| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | "net/http" |
| 8 | "net/url" |
| 9 | |
| 10 | "reasonix/internal/agent" |
| 11 | "reasonix/internal/sessioninbox" |
| 12 | ) |
| 13 | |
| 14 | // remoteInboxSnapshot reads the selected durable queue using the same route |
| 15 | // identity as remote runtime synchronization. A reply from a replaced tunnel |
| 16 | // or selection must not clear or populate the current composer's queue. |
| 17 | func (a *App) remoteInboxSnapshot(tabID string) (InboxSnapshotView, error) { |
| 18 | a.remoteTabMu.Lock() |
| 19 | tab := a.remoteTabs[tabID] |
| 20 | if tab == nil || tab.client == nil || tab.state != "ready" || tab.routing.currentPath == "" || tab.routing.rehydratingPath != "" { |
| 21 | a.remoteTabMu.Unlock() |
| 22 | return InboxSnapshotView{}, fmt.Errorf("remote tab %q is not ready for inbox reads", tabID) |
| 23 | } |
| 24 | client, base, path := tab.client, tab.base, tab.routing.currentPath |
| 25 | gen, selection := tab.gen, tab.selectionRevision |
| 26 | a.remoteTabMu.Unlock() |
| 27 | |
| 28 | ctx, cancel := commandContext(a) |
| 29 | defer cancel() |
| 30 | data, err := serveGet(ctx, client, serveURL(base, "/inbox?session="+url.QueryEscape(path))) |
| 31 | if err != nil { |
| 32 | return InboxSnapshotView{}, err |
| 33 | } |
| 34 | var snap sessioninbox.InboxSnapshot |
| 35 | if err := json.Unmarshal(data, &snap); err != nil { |
| 36 | return InboxSnapshotView{}, err |
| 37 | } |
| 38 | if snap.SessionPath == "" || agent.CanonicalSessionPath(snap.SessionPath) != agent.CanonicalSessionPath(path) { |
| 39 | return InboxSnapshotView{}, fmt.Errorf("remote inbox session changed") |
| 40 | } |
| 41 | a.remoteTabMu.Lock() |
| 42 | defer a.remoteTabMu.Unlock() |
| 43 | current := a.remoteTabs[tabID] |
| 44 | if current != tab || current.gen != gen || current.selectionRevision != selection || current.client != client || current.base != base || current.state != "ready" || current.routing.currentPath != path || current.routing.rehydratingPath != "" { |
| 45 | return InboxSnapshotView{}, fmt.Errorf("remote inbox route changed during read") |
| 46 | } |
| 47 | return inboxSnapshotView(snap), nil |
| 48 | } |
| 49 | |
| 50 | // enqueueRemoteFollowup preserves the route, rich input and caller's stable |
| 51 | // key. An uncertain POST is reconciled by reading its receipt, never replayed. |
| 52 | func (a *App) enqueueRemoteFollowup(tabID, display, submit string, invocations []InvocationRequest, idempotency string) (InboxReceiptView, error) { |
| 53 | if err := a.requireRemoteExecutionProtocol(tabID); err != nil { |
| 54 | return InboxReceiptView{}, err |
| 55 | } |
| 56 | client, base, path, err := a.remoteTabCommandTarget(tabID) |
| 57 | if err != nil { |
| 58 | return InboxReceiptView{}, err |
| 59 | } |
| 60 | return a.enqueueRemoteFollowupAt(client, base, path, display, submit, invocations, idempotency) |
| 61 | } |
| 62 | |
| 63 | func (a *App) enqueueRemoteFollowupAt(client *http.Client, base, path, display, submit string, invocations []InvocationRequest, idempotency string) (InboxReceiptView, error) { |
| 64 | ctx, cancel := commandContext(a) |
| 65 | defer cancel() |
| 66 | body, err := json.Marshal(map[string]any{"intent": "followup", "input": submit, "display": display, "invocations": invocations, "idempotencyKey": idempotency}) |
| 67 | if err != nil { |
| 68 | return InboxReceiptView{}, err |
| 69 | } |
| 70 | resp, err := serveDoForSession(ctx, client, http.MethodPost, serveURL(base, "/inbox/items"), body, path) |
| 71 | if err == nil { |
| 72 | defer resp.Body.Close() |
| 73 | if resp.StatusCode >= 200 && resp.StatusCode < 300 { |
| 74 | var receipt InboxReceiptView |
| 75 | err = json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&receipt) |
| 76 | if err == nil && receipt.ItemID != "" { |
| 77 | return receipt, nil |
| 78 | } |
| 79 | } else { |
| 80 | err = fmt.Errorf("follow-up enqueue failed (%d)", resp.StatusCode) |
| 81 | if resp.StatusCode >= 400 && resp.StatusCode < 500 { |
| 82 | if resp.StatusCode != http.StatusRequestTimeout { |
| 83 | return InboxReceiptView{}, inboxNotSubmitted(err) |
| 84 | } |
| 85 | return InboxReceiptView{}, err |
| 86 | } |
| 87 | } |
| 88 | } |
| 89 | if idempotency != "" { |
| 90 | lookupCtx, lookupCancel := commandContext(a) |
| 91 | defer lookupCancel() |
| 92 | data, lookupErr := serveGet(lookupCtx, client, serveURL(base, "/inbox/receipt?key="+url.QueryEscape(idempotency)+"&session="+url.QueryEscape(path))) |
| 93 | var receipt InboxReceiptView |
| 94 | if lookupErr == nil && json.Unmarshal(data, &receipt) == nil && receipt.ItemID != "" { |
| 95 | return receipt, nil |
| 96 | } |
| 97 | } |
| 98 | if err == nil { |
| 99 | err = fmt.Errorf("follow-up receipt unavailable") |
| 100 | } |
| 101 | return InboxReceiptView{}, err |
| 102 | } |
| 103 |