From 6e1eebd53599dd3d6077fa0f380c4f97fbb847f6 Mon Sep 17 00:00:00 2001 From: Matt Date: Tue, 28 Jul 2026 20:06:15 -0400 Subject: [PATCH 1/2] feat(compass): SEA-1364 T3 commit relayed agent conversation frames to comms Replaces the observedConversationSink stub with the real comms write-through: a relayed conversation frame now commits a durable Message row under the agent's resolved account instead of being acked and discarded. The Server half is complete; the end-to-end path is still dead and the missing piece is not in this lane. The agent's only conversation emitter never stamps a message id (Message.id is server-assigned by contract), and nothing between the agent and the hub stamps one, so every real frame is an id-less update that addresses no row. Until the Runner reconciles server-minted ids, zero conversation rows commit. Rather than hide that, it is pinned: id-less frames get their own contract-defect classification and counter, distinct from ordinary per-frame refusals, logged at ERROR stating the consequence; a test drives the exact shape the agent emits and asserts today's real behavior; counters are exported and logged on drain. Also: a nil-deref panic in homeChannel for a non-agent account, a silently dropped parent_message_id, and an impossible-state error mapped into the routine-refusal bucket. Co-Authored-By: seal --- go/internal/board/deliver_integration_test.go | 3 +- go/internal/comms/agent_caller.go | 163 +++++- .../comms/agent_conversation_pgtest_test.go | 501 ++++++++++++++++++ go/internal/runnerhub/contract_defect_test.go | 270 ++++++++++ .../conversation_writethrough_test.go | 198 +++++++ go/internal/runnerhub/helpers_test.go | 62 ++- go/internal/runnerhub/hub.go | 276 +++++++++- go/internal/runnerhub/hub_test.go | 16 +- .../runnerhub/integration_pgtest_test.go | 2 +- go/internal/runnerhub/seam_test.go | 8 + go/internal/store/messages.go | 90 ++++ .../store/messages_authored_update_test.go | 253 +++++++++ go/server/serve.go | 103 ++-- go/server/sinks.go | 151 ++++-- go/server/sinks_test.go | 56 ++ 15 files changed, 2060 insertions(+), 92 deletions(-) create mode 100644 go/internal/comms/agent_conversation_pgtest_test.go create mode 100644 go/internal/runnerhub/contract_defect_test.go create mode 100644 go/internal/runnerhub/conversation_writethrough_test.go create mode 100644 go/internal/store/messages_authored_update_test.go create mode 100644 go/server/sinks_test.go diff --git a/go/internal/board/deliver_integration_test.go b/go/internal/board/deliver_integration_test.go index a4a40e23..e0b42f0e 100644 --- a/go/internal/board/deliver_integration_test.go +++ b/go/internal/board/deliver_integration_test.go @@ -20,6 +20,7 @@ import ( compassv1 "github.com/sealedsecurity/compass/go/gen/compass/v1" compassv1internal "github.com/sealedsecurity/compass/go/internal/gen/compass/v1" "github.com/sealedsecurity/compass/go/internal/runnerhub" + "github.com/sealedsecurity/compass/go/internal/store" ) // noopConversationSink is a do-nothing ConversationSink: this seam test targets @@ -27,7 +28,7 @@ import ( // nothing. type noopConversationSink struct{} -func (noopConversationSink) PostAgentMessage(_ context.Context, _ string, _ *compassv1.MessagePosted, _ *compassv1.MessageUpdated) error { +func (noopConversationSink) PostAgentMessage(_ context.Context, _ store.AccountID, _ string, _ *compassv1.MessagePosted, _ *compassv1.MessageUpdated) error { return nil } diff --git a/go/internal/comms/agent_caller.go b/go/internal/comms/agent_caller.go index 8a93962b..0d96e69c 100644 --- a/go/internal/comms/agent_caller.go +++ b/go/internal/comms/agent_caller.go @@ -1,11 +1,18 @@ //go:build unix -// The agent-comms execution leg: PostAsAccount / ListAsAccount execute one -// agent-initiated comms call as a resolved agent account. The RunnerHub's -// RelayCommsCall handler (internal/runnerhub) calls these after resolving the -// relayed session_id to its bound account — the Runner asserts no account, the -// Server attributes in-process here (transport design Decision #3 / OQ-2, -// comms-tools design T2). +// The agent-comms *AsAccount family: execute one agent-originated comms +// operation as a resolved agent account. Two callers, both in the RunnerHub +// (internal/runnerhub), both resolving the relayed session_id to its bound +// account first — the Runner asserts no account, the Server attributes +// in-process here (transport design Decision #3 / OQ-2, comms-tools design T2): +// +// - PostAsAccount / ListAsAccount serve RelayCommsCall — a comms call the +// agent made deliberately, as a tool. +// - CommitAgentPost / CommitAgentUpdate serve the ConversationSink — the +// write-through that turns a relayed conversation FRAME (the agent's own +// turn, streamed out as it speaks) into a durable comms row (SEA-1364 T3). +// They are built on the first pair and the authorizing store update rather +// than being a second write path. // // These are deliberately NOT new CommsService RPCs: an agent-initiated call // never reaches a network door (it rides the per-container socket to the Runner, @@ -47,6 +54,31 @@ func (noActorError) Error() string { return "comms: agent-initiated call requires a resolved account; refusing to attribute to the bootstrap admin" } +// errNotAgentAccount is the fail-closed cause when an agent-initiated call +// resolves to an account that is NOT an agent — a user account, whose +// Account.Agent is nil (scanAccount sets the subtype only for the side of the +// join that populated, store/accounts.go:313-322). Only an agent account has a +// home channel, so there is no channel to default to and nothing to post into. +// +// It maps to CodeFailedPrecondition, NOT CodeInvalidArgument, and the difference +// is load-bearing at the relay. The RunnerHub classifies a sink error by its +// Connect code: InvalidArgument is a routine per-frame refusal, counted among +// the frames a healthy relay occasionally rejects, while FailedPrecondition is a +// CONTRACT DEFECT — counted separately and logged as a misconfiguration +// (runnerhub/hub.go, isContractDefect). A session bound to a user account is the +// latter: no retry and no better-formed frame can fix it, and every frame from +// that session will fail identically, so it must not read as routine garbage. +var errNotAgentAccount = connect.NewError( + connect.CodeFailedPrecondition, + notAgentAccountError{}, +) + +type notAgentAccountError struct{} + +func (notAgentAccountError) Error() string { + return "comms: agent-initiated call resolved to a non-agent account, which has no home channel; the session->account binding is wrong" +} + // PostAsAccount executes one agent-initiated PostMessage as account. It sets the // account on the context (WithActor) and delegates to the same PostMessage // handler path a human caller takes, so authz (D9), idempotency @@ -102,6 +134,101 @@ func (c *Comms) ListAsAccount( return resp.Msg, nil } +// CommitAgentPost commits one relayed MessagePosted frame as a durable comms row +// under account — the write-through the RunnerHub's ConversationSink drives +// (SEA-1364 T3). It builds a PostMessageRequest from the frame's blocks and +// delegates to PostAsAccount, so this is the SAME PostMessage handler path a +// human takes: same store calls, same D9 write-authz, same idempotency, same +// MessagePosted fan-out. No new authz code exists here to drift from the human +// path. +// +// The Container oneof is deliberately left UNSET. defaultChannel (:109) fills an +// empty channel_id from the account's home channel, so a relayed frame routes to +// the agent's own channel for free — the agent mints no server ids and the +// Runner plumbs no channel id, which is exactly the contract (the frame's +// Message.id / channel are server-assigned, comms.proto:234-242, so any value +// the frame carried would be meaningless here). +// +// parent_message_id IS threaded, unlike the two fields above. It is plumbed end +// to end on both the wire Message (comms.proto:250) and PostMessageRequest +// (comms.proto:565), and it is the ONE piece of routing the frame legitimately +// carries: the id it names is a SERVER id, from a row that already committed, so +// forwarding it asserts nothing the agent minted. Dropping it would silently +// flatten every threaded agent reply to a root message — no error, no log, just +// a conversation that lost its shape. The store validates the parent exists, and +// AppendMessage's membership gate applies to it exactly as it does for a human +// reply, so a frame naming a parent it may not see is refused, not honored. +// +// Idempotency: ClientRequestId is left unset because the relayed frame carries +// no key on this base. PublishEventsRequest is {runner_seq, session_id, frame} +// (proto/compass/v1/runner.proto:169-183), and the agent-minted key lives one hop +// earlier on PostConversationFrameRequest.idempotency_key +// (proto/compass/v1/agent_gateway.proto:113-120), where it terminates at the +// Runner's C2 dedup and is not forwarded upstream. This is NOT a deliberate +// decision to go unkeyed: #894/T2 adds idempotency_key to PublishEventsRequest +// and threads it to this seam, at which point it populates ClientRequestId below +// and the store's (author_account_id, client_request_id) unique constraint +// (messages.go:82) makes the commit genuinely at-most-once. Until then a retried +// frame would commit twice — the gap is the missing key, and the hook is the one +// field on the request built here. +func (c *Comms) CommitAgentPost( + ctx context.Context, + account store.AccountID, + posted *compassv1.MessagePosted, +) (*compassv1.PostMessageResponse, error) { + if account == "" { + return nil, errNoActor + } + return c.PostAsAccount(ctx, account, &compassv1.PostMessageRequest{ + // Container unset on purpose — see the home-channel note above. + Blocks: posted.GetMessage().GetBlocks(), + ParentMessageId: posted.GetMessage().GetParentMessageId(), + // ClientRequestId: threaded from the frame's idempotency_key by #894/T2. + }) +} + +// CommitAgentUpdate applies one relayed MessageUpdated frame to the row it +// addresses, under account. A streaming turn re-sends its FULL current block set +// (comms.proto:388-390), so this replaces the block set rather than appending. +// +// It goes through store.UpdateMessageBlocksAsAuthor rather than the store's +// unauthorizing core: that core takes a bare MessageID with no membership and no +// authorship check, which is safe for AnswerAsk's already-gated call but is a +// privilege hole for an id that arrived on a relayed frame. The authorizing +// variant requires the actor to be both a member of the message's channel and its +// author, and collapses every refusal — unknown id, foreign message, revoked +// membership — to one ErrNotFound, so a relayed frame cannot enumerate messages +// it may not touch (the D9 not-found/forbidden merge). +// +// Errors map through edgeError to the same Connect codes a human caller gets: +// CodeNotFound for a refused row, CodeInvalidArgument for a malformed frame (an +// empty message.id, an empty block set, an ask block missing its immutable +// ask_id). Those are the refusals the hub treats as non-fatal drops rather than +// stream teardowns. +func (c *Comms) CommitAgentUpdate( + ctx context.Context, + account store.AccountID, + updated *compassv1.MessageUpdated, +) (*compassv1.MessageUpdated, error) { + if account == "" { + return nil, errNoActor + } + msg := updated.GetMessage() + blocks, err := blocksFromWire(msg.GetBlocks()) + if err != nil { + return nil, err + } + stored, err := c.store.UpdateMessageBlocksAsAuthor(ctx, account, store.MessageID(msg.GetId()), blocks) + if err != nil { + return nil, edgeError(err) + } + // Write first, then fan out — the same write-through order PostMessage and + // RespondToAsk use, so a subscriber never sees an event for a row that did + // not commit. + c.publishMessageUpdated(stored) + return &compassv1.MessageUpdated{Message: messageToWire(stored)}, nil +} + // defaultChannel returns req with an empty channel_id filled from the account's // home channel, so the common "post in my own channel" case needs no id plumbed // into the container. A non-empty channel_id is left untouched. The returned @@ -135,14 +262,30 @@ func (c *Comms) defaultListChannel(ctx context.Context, account store.AccountID, return c.homeChannel(ctx, account) } -// homeChannel resolves the account's home channel id. The account must be an -// agent account (it always has a home channel, minted at CreateAgent — -// store/accounts.go:156-158); a store failure surfaces mapped to its Connect -// code, never leaked verbatim. +// homeChannel resolves the account's home channel id. An agent account always +// has one, minted at CreateAgent (store/accounts.go:156-158); a store failure +// surfaces mapped to its Connect code, never leaked verbatim. +// +// A NON-AGENT account is refused rather than dereferenced. scanAccount populates +// Account.Agent only for the agent side of the join (store/accounts.go:313-322), +// so for a user account it is nil and reading acc.Agent.HomeChannelID panicked — +// inside the RunnerHub's PublishEvents handler goroutine, where a panic takes the +// relay down and no error ever reaches the classification the hub does on the +// return value (rule://go-no-panic-in-lib). Failing closed with +// errNotAgentAccount both keeps the process up and gives the hub a code it can +// route: FailedPrecondition marks it a contract defect, so a bad binding is +// reported as the wiring fault it is instead of hiding among per-frame refusals. +// +// This is the call-site guard only. The deeper question — that a container can be +// bound to a non-agent account at all, i.e. the ordering of bindContainer's +// validation against RecordAgentContainer — belongs to that flow, not here. func (c *Comms) homeChannel(ctx context.Context, account store.AccountID) (string, error) { acc, err := c.store.GetAccount(ctx, account) if err != nil { return "", edgeError(err) } + if acc.Agent == nil { + return "", errNotAgentAccount + } return string(acc.Agent.HomeChannelID), nil } diff --git a/go/internal/comms/agent_conversation_pgtest_test.go b/go/internal/comms/agent_conversation_pgtest_test.go new file mode 100644 index 00000000..ea260c1d --- /dev/null +++ b/go/internal/comms/agent_conversation_pgtest_test.go @@ -0,0 +1,501 @@ +//go:build pgtest && unix + +package comms + +// The conversation write-through (SEA-1364 T3): CommitAgentPost / CommitAgentUpdate +// turn a relayed agent conversation frame into a durable comms row under the +// account the RunnerHub resolved the session to. Every test drives the real +// Postgres store + real bus (the newHandler / newStreamHarness harnesses the rest +// of this package uses — no mocks) and defends one contract: +// +// - a posted frame lands on the agent's HOME channel, attributed to the AGENT +// account, and fans out on SubscribeComms — indistinguishable downstream from +// a human post, because it IS the PostMessage handler path; +// - an updated frame edits the row in place through the AUTHORIZING store +// update, and fans out as MessageUpdated; +// - a cross-account update, an update from a revoked member, an ask block with +// no id, and an empty message.id are each REFUSED with the code a human +// caller would get — the refusals the hub then treats as non-fatal; +// - an empty account is a hard CodeInvalidArgument on both methods, never a +// silent fall-through to bootstrap-admin attribution. +// +// Gated `pgtest && unix` like agent_caller_pgtest_test.go: it SKIPs (via +// pgtest.RequireDSN in newTestStore) when no Postgres/podman runtime exists, and +// `unix` because agent_caller.go is unix-tagged. + +import ( + "context" + "testing" + + "connectrpc.com/connect" + + compassv1 "github.com/sealedsecurity/compass/go/gen/compass/v1" + "github.com/sealedsecurity/compass/go/internal/store" +) + +// postedFrame wraps blocks in the MessagePosted frame shape the Runner relays. +// The Container oneof is deliberately left UNSET — that is what routes the +// commit to the agent's home channel (defaultChannel, agent_caller.go:109). +func postedFrame(blocks []*compassv1.MessageBlock) *compassv1.MessagePosted { + return &compassv1.MessagePosted{Message: &compassv1.Message{Blocks: blocks}} +} + +// updatedFrame wraps an addressed message id + blocks in the MessageUpdated +// frame shape. Unlike a post, an update MUST carry the row's id. +func updatedFrame(id string, blocks []*compassv1.MessageBlock) *compassv1.MessageUpdated { + return &compassv1.MessageUpdated{Message: &compassv1.Message{Id: id, Blocks: blocks}} +} + +// askBlockWire builds a wire ask block carrying one question, for the malformed +// -ask rejection case (the id an update must carry is the store's, minted at +// append; a wire ask always enters id-less, so an UPDATE carrying one is +// necessarily missing its immutable id). +func askBlockWire() *compassv1.MessageBlock { + return &compassv1.MessageBlock{Block: &compassv1.MessageBlock_Ask{Ask: &compassv1.Ask{ + Questions: []*compassv1.AskQuestion{{ + QuestionId: "q1", + Question: "Which environment?", + Options: []*compassv1.AskOption{ + {Id: "opt-a", Label: "staging"}, + {Id: "opt-b", Label: "prod"}, + }, + }}, + }}} +} + +// A relayed posted frame becomes a real Message row on the agent's HOME channel, +// authored by the AGENT account. This is the core dogfood contract the stub +// broke: the frame was acked as committed and then discarded, so nothing landed. +// +// Mutation that reddens it: dropping the AppendMessage delegation (back to a log +// line) → the read-back finds no row; hardcoding an actor → the author assertion +// fails. +func TestCommitAgentPostLandsOnHomeChannelAsTheAgent(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + + resp, err := svc.CommitAgentPost(ctx, agent.ID, postedFrame(textBlocks("the agent speaks"))) + if err != nil { + t.Fatalf("CommitAgentPost: %v", err) + } + if got := resp.GetMessage().GetChannelId(); got != string(agent.Agent.HomeChannelID) { + t.Fatalf("committed message channel = %q, want the agent's home channel %q (the Container oneof must be left unset)", got, agent.Agent.HomeChannelID) + } + if got := resp.GetMessage().GetAuthorAccountId(); got != string(agent.ID) { + t.Fatalf("committed message author = %q, want the agent account %q (never the bootstrap admin)", got, agent.ID) + } + + // Durable: the row is readable back out of the store of record. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages, want exactly 1 (the relayed frame must COMMIT, not be observed)", len(msgs)) + } + if msgs[0].ID != store.MessageID(resp.GetMessage().GetId()) { + t.Fatalf("stored message id = %q, want the returned %q", msgs[0].ID, resp.GetMessage().GetId()) + } +} + +// The fan-out half: a committed post reaches a live SubscribeComms subscriber as +// MessagePosted, so a relayed agent turn is visible to a watching client exactly +// as a human post is. Driven over the real stream (newStreamHarness) with the +// subscribe running concurrently with the commit — connect server-streaming over +// HTTP/1 is half-duplex, so the subscribe must not precede the mutation. +func TestCommitAgentPostFansOutOnSubscribeComms(t *testing.T) { + h := newStreamHarness(t) + ctx := context.Background() + + owner := mustUser(t, h.store, "owner") + agent := mustAgent(t, h.store, owner.ID, "agent") + + // Subscribe AS the agent: it is a member of its own home channel, so the D9 + // stream filter passes the event to it. + events := firstEventAfterBoundary(t, h, agent.ID, &compassv1.SubscribeCommsRequest{SinceSeq: 0}) + + resp, err := h.svc.CommitAgentPost(ctx, agent.ID, postedFrame(textBlocks("fanned out"))) + if err != nil { + t.Fatalf("CommitAgentPost: %v", err) + } + + got := awaitFirst(t, events) + posted := got.GetMessagePosted() + if posted == nil { + t.Fatalf("stream event payload = %T, want a MessagePosted", got.GetPayload()) + } + if posted.GetMessage().GetId() != resp.GetMessage().GetId() { + t.Fatalf("fanned message id = %q, want the committed %q", posted.GetMessage().GetId(), resp.GetMessage().GetId()) + } + if got := posted.GetMessage().GetAuthorAccountId(); got != string(agent.ID) { + t.Fatalf("fanned message author = %q, want the agent account %q", got, agent.ID) + } +} + +// A relayed updated frame edits the addressed row IN PLACE — same id, new blocks, +// no second row — and fans out as MessageUpdated. This is the streaming-turn +// path: the agent appends a block and re-sends the full current set. +func TestCommitAgentUpdateEditsInPlaceAndFansOut(t *testing.T) { + h := newStreamHarness(t) + ctx := context.Background() + + owner := mustUser(t, h.store, "owner") + agent := mustAgent(t, h.store, owner.ID, "agent") + + // Seed the row to be edited through the STORE, not CommitAgentPost: a post + // publishes MessagePosted onto the bus, and a since_seq=0 subscribe replays + // it, so the update's own fan-out would not be the first event. Appending + // directly writes the row without an event, isolating the assertion to the + // MessageUpdated this test is about. + seedText := "partial" + seed, _, err := h.store.AppendMessage(ctx, store.Message{ + Container: store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, + AuthorAccountID: agent.ID, + Blocks: []store.MessageBlock{{Text: &seedText}}, + }, "") + if err != nil { + t.Fatalf("AppendMessage(seed): %v", err) + } + id := string(seed.ID) + + events := firstEventAfterBoundary(t, h, agent.ID, &compassv1.SubscribeCommsRequest{SinceSeq: 0}) + + updated, err := h.svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, textBlocks("partial and settled"))) + if err != nil { + t.Fatalf("CommitAgentUpdate: %v", err) + } + if updated.GetMessage().GetId() != id { + t.Fatalf("updated message id = %q, want the edited row %q (an update must not insert)", updated.GetMessage().GetId(), id) + } + + got := awaitFirst(t, events) + if ev := got.GetMessageUpdated(); ev == nil { + t.Fatalf("stream event payload = %T, want a MessageUpdated", got.GetPayload()) + } else if ev.GetMessage().GetId() != id { + t.Fatalf("fanned updated id = %q, want %q", ev.GetMessage().GetId(), id) + } + + // Exactly one row, carrying the new text. + msgs, err := h.store.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages after an update, want exactly 1 (in-place edit)", len(msgs)) + } + if msgs[0].Blocks[0].Text == nil || *msgs[0].Blocks[0].Text != "partial and settled" { + t.Fatalf("stored text = %v, want the updated body", msgs[0].Blocks[0].Text) + } +} + +// THE cross-account security case at the comms seam: agent B cannot edit agent +// A's message, even in a channel B can read. It collapses to CodeNotFound — the +// same answer B gets for a message that does not exist — so B cannot enumerate +// A's messages by id. The positive control (A editing its own row) keeps this +// from passing vacuously. +// +// Mutation: route this through updateMessageBlocksExec (the bare-id, no-authz +// core this fork exists to avoid) → B's edit succeeds and BOTH the error and the +// untouched-content assertions redden. +func TestCommitAgentUpdateCrossAccountIsNotFound(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agentA := mustAgent(t, st, owner.ID, "agent-a") + agentB := mustAgent(t, st, owner.ID, "agent-b") + + // A shared channel both agents belong to, so B can READ A's message: only + // authorship refuses the edit. + ch, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{ + Name: "shared", Kind: store.ChannelKindChannel, + MemberAccountIDs: []store.AccountID{agentA.ID, agentB.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + postedA, err := svc.PostAsAccount(ctx, agentA.ID, &compassv1.PostMessageRequest{ + Container: &compassv1.PostMessageRequest_ChannelId{ChannelId: string(ch.ID)}, + Blocks: textBlocks("agent A's words"), + }) + if err != nil { + t.Fatalf("PostAsAccount(A): %v", err) + } + id := postedA.GetMessage().GetId() + + _, err = svc.CommitAgentUpdate(ctx, agentB.ID, updatedFrame(id, textBlocks("put into A's mouth"))) + connectCodeIs(t, err, connect.CodeNotFound, "CommitAgentUpdate(cross-account)") + + // Positive control: A can edit its own row, so B's refusal is authorship. + if _, err := svc.CommitAgentUpdate(ctx, agentA.ID, updatedFrame(id, textBlocks("agent A's own edit"))); err != nil { + t.Fatalf("agent A cannot edit its own message: %v", err) + } + + msgs, err := st.ListMessages(ctx, agentA.ID, store.ContainerRef{ChannelID: ch.ID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if msgs[0].Blocks[0].Text == nil || *msgs[0].Blocks[0].Text != "agent A's own edit" { + t.Fatalf("stored text = %v, want A's own edit (B's rejected update must leave no trace)", msgs[0].Blocks[0].Text) + } +} + +// The membership half at the comms seam: an agent removed from the channel it +// posted in can no longer edit its own past message — CodeNotFound, so write +// access dies with read access. A bare authorship check would let this through. +func TestCommitAgentUpdateRevokedMemberIsNotFound(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + ch, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{ + Name: "room", Kind: store.ChannelKindChannel, + MemberAccountIDs: []store.AccountID{agent.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + posted, err := svc.PostAsAccount(ctx, agent.ID, &compassv1.PostMessageRequest{ + Container: &compassv1.PostMessageRequest_ChannelId{ChannelId: string(ch.ID)}, + Blocks: textBlocks("posted while a member"), + }) + if err != nil { + t.Fatalf("PostAsAccount: %v", err) + } + id := posted.GetMessage().GetId() + + // Positive control before the revoke. + if _, err := svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, textBlocks("edited while a member"))); err != nil { + t.Fatalf("the agent cannot edit while still a member: %v", err) + } + + if _, _, err := st.UpdateChannelMembers(ctx, owner.ID, ch.ID, []store.MemberUpdate{{AccountID: agent.ID, Remove: true}}); err != nil { + t.Fatalf("UpdateChannelMembers(remove): %v", err) + } + + _, err = svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, textBlocks("edited after the revoke"))) + connectCodeIs(t, err, connect.CodeNotFound, "CommitAgentUpdate(revoked member)") + + msgs, err := st.ListMessages(ctx, owner.ID, store.ContainerRef{ChannelID: ch.ID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if msgs[0].Blocks[0].Text == nil || *msgs[0].Blocks[0].Text != "edited while a member" { + t.Fatalf("stored text = %v, want the pre-revoke text", msgs[0].Blocks[0].Text) + } +} + +// The two malformed-input refusals, both CodeInvalidArgument: an update whose +// message.id is empty (a frame the agent never stamped) and an update carrying +// an ask block with no ask_id (which would orphan a pending RespondToAsk). +func TestCommitAgentUpdateRejectsMalformedFrames(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + posted, err := svc.CommitAgentPost(ctx, agent.ID, postedFrame(textBlocks("untouched"))) + if err != nil { + t.Fatalf("CommitAgentPost: %v", err) + } + + t.Run("empty message id", func(t *testing.T) { + _, err := svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame("", textBlocks("nowhere to land"))) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(empty message id)") + }) + t.Run("ask block with no ask id", func(t *testing.T) { + // A wire ask always enters id-less (askFromWire strips any caller value), + // so an UPDATE carrying one is necessarily missing the immutable id the + // store minted at append. + _, err := svc.CommitAgentUpdate(ctx, agent.ID, + updatedFrame(posted.GetMessage().GetId(), []*compassv1.MessageBlock{askBlockWire()})) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(ask with no ask_id)") + }) + + // Neither refusal wrote anything: the original row is intact. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages, want 1", len(msgs)) + } + if msgs[0].Blocks[0].Text == nil || *msgs[0].Blocks[0].Text != "untouched" { + t.Fatalf("stored text = %v, want the untouched original", msgs[0].Blocks[0].Text) + } +} + +// Fail-closed identity on BOTH write-through methods, mirroring +// TestPostAsAccountEmptyAccountFailsClosedNoAdminWrite for this leg: an empty +// resolved account is a hard CodeInvalidArgument and writes NOTHING. Without the +// guard, actorFromContext's admin fallback (comms.go:331-336) would attribute the +// agent's words to the bootstrap admin — the exact silent misattribution the +// fail-closed posture exists to prevent. +func TestCommitAgentFramesEmptyAccountFailsClosed(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + seeded, err := svc.CommitAgentPost(ctx, agent.ID, postedFrame(textBlocks("the agent's own"))) + if err != nil { + t.Fatalf("CommitAgentPost(seed): %v", err) + } + + _, err = svc.CommitAgentPost(ctx, "", postedFrame(textBlocks("should never be written"))) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentPost(empty account)") + + _, err = svc.CommitAgentUpdate(ctx, "", updatedFrame(seeded.GetMessage().GetId(), textBlocks("should never be applied"))) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(empty account)") + + // Nothing was written or altered under any fallback identity. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages, want exactly 1 (the unresolved calls must write nothing)", len(msgs)) + } + if msgs[0].Blocks[0].Text == nil || *msgs[0].Blocks[0].Text != "the agent's own" { + t.Fatalf("stored text = %v, want the agent's own seeded text", msgs[0].Blocks[0].Text) + } +} + +// Idempotent dedup at the seam the relayed key WILL enter. The relayed frame +// carries no idempotency key on this base — PublishEventsRequest is +// {runner_seq, session_id, frame} (runner.proto:169-183) and the agent-minted +// key terminates at the Runner (agent_gateway.proto:113-120) — so the +// write-through leaves ClientRequestId unset and this contract cannot be +// exercised THROUGH a relayed frame yet. It is pinned here, on the +// PostMessageRequest the write-through builds and delegates, because that is the +// exact field #894/T2 populates: when the key arrives, this test is already the +// proof that a retry commits no second row. +func TestCommitAgentPostRequestIsDedupedByClientRequestID(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + + // The request shape CommitAgentPost builds (home-channel default via an + // unset Container), plus the key T2 will thread onto it. + post := func(text string) (*compassv1.PostMessageResponse, error) { + return svc.PostAsAccount(ctx, agent.ID, &compassv1.PostMessageRequest{ + Blocks: textBlocks(text), + ClientRequestId: "relayed-frame-1", + }) + } + first, err := post("the turn") + if err != nil { + t.Fatalf("first commit: %v", err) + } + retry, err := post("the turn") + if err != nil { + t.Fatalf("retry commit: %v", err) + } + if first.GetMessage().GetId() != retry.GetMessage().GetId() { + t.Fatalf("retry stored a new message %q, want the first's %q (dedup on (author, client_request_id))", retry.GetMessage().GetId(), first.GetMessage().GetId()) + } + + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages after a keyed retry, want exactly 1", len(msgs)) + } +} + +// A relayed frame that names a parent keeps it: the committed row is threaded +// under the message it replies to. parent_message_id is plumbed end to end on +// both the wire Message (comms.proto:250) and PostMessageRequest +// (comms.proto:565), so the write-through dropping it would silently flatten +// every threaded agent reply to a root message — a data loss that no error +// reports and that reads, downstream, as the agent simply never having replied +// in-thread. +// +// Mutation: drop ParentMessageId from the request CommitAgentPost builds → the +// committed message comes back with an empty parent and this reddens twice (the +// response echo and the store read-back). +func TestCommitAgentPostThreadsTheFramesParentMessageID(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + + root, err := svc.CommitAgentPost(ctx, agent.ID, postedFrame(textBlocks("the question"))) + if err != nil { + t.Fatalf("CommitAgentPost(root): %v", err) + } + rootID := root.GetMessage().GetId() + if got := root.GetMessage().GetParentMessageId(); got != "" { + t.Fatalf("root committed message ParentMessageId = %q, want \"\" (a frame with no parent must stay a root)", got) + } + + // The frame the agent relays for an in-thread reply: same shape as any + // posted frame, plus the parent it answers. + reply := postedFrame(textBlocks("the threaded answer")) + reply.Message.ParentMessageId = rootID + + committed, err := svc.CommitAgentPost(ctx, agent.ID, reply) + if err != nil { + t.Fatalf("CommitAgentPost(threaded reply): %v", err) + } + if got := committed.GetMessage().GetParentMessageId(); got != rootID { + t.Fatalf("committed reply ParentMessageId = %q, want the root %q — the relayed frame's parent must survive the write-through", got, rootID) + } + + // Durable, not merely echoed: the store of record holds the threading. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + parents := map[store.MessageID]store.MessageID{} + for _, m := range msgs { + parents[m.ID] = m.ParentMessageID + } + if got := parents[store.MessageID(committed.GetMessage().GetId())]; got != store.MessageID(rootID) { + t.Fatalf("stored reply ParentMessageID = %q, want the root %q", got, rootID) + } + if got := parents[store.MessageID(rootID)]; got != "" { + t.Fatalf("stored root ParentMessageID = %q, want \"\"", got) + } +} + +// A session bound to a NON-AGENT account is refused, not a panic. homeChannel +// reads acc.Agent.HomeChannelID, and scanAccount (store/accounts.go:313-322) sets +// acc.Agent only for agent accounts — for a user account it is nil, so before +// this guard the deref panicked inside the PublishEvents handler goroutine, +// taking the relay down with it (rule://go-no-panic-in-lib). +// +// It is a CONTRACT DEFECT rather than an ordinary refusal, hence +// CodeFailedPrecondition: a user-account binding is not a request the caller +// could pose differently, it is the hub and the store disagreeing about what a +// session resolves to. The hub counts it separately and logs it as a +// misconfiguration (runnerhub/hub.go, isContractDefect) instead of burying it +// among expected per-frame refusals. +// +// Both write-through methods and both *AsAccount methods are covered: every one +// of them reaches homeChannel through the empty-container/empty-channel default, +// so a guard on only one call site would leave the panic reachable. +func TestAgentCallsWithANonAgentAccountAreRefusedNotPanics(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + // A plain USER account: real and resolvable, but acc.Agent is nil. + user := mustUser(t, st, "human") + + _, err := svc.CommitAgentPost(ctx, user.ID, postedFrame(textBlocks("never committed"))) + connectCodeIs(t, err, connect.CodeFailedPrecondition, "CommitAgentPost(user account)") + + _, err = svc.PostAsAccount(ctx, user.ID, &compassv1.PostMessageRequest{Blocks: textBlocks("never committed")}) + connectCodeIs(t, err, connect.CodeFailedPrecondition, "PostAsAccount(user account, empty channel)") + + _, err = svc.ListAsAccount(ctx, user.ID, &compassv1.ListMessagesRequest{}) + connectCodeIs(t, err, connect.CodeFailedPrecondition, "ListAsAccount(user account, empty channel)") +} diff --git a/go/internal/runnerhub/contract_defect_test.go b/go/internal/runnerhub/contract_defect_test.go new file mode 100644 index 00000000..a75f5060 --- /dev/null +++ b/go/internal/runnerhub/contract_defect_test.go @@ -0,0 +1,270 @@ +//go:build unix + +package runnerhub + +// The CONTRACT-DEFECT classification (SEA-1364 T3 review): the frame class that +// is not one bad frame but a broken relay. Deliver already had two buckets — a +// per-frame refusal (drop, count, carry on) and an infrastructure fault (tear the +// stream down so the Runner retries). Neither fits a frame the agent and the +// Server disagree about the SHAPE of, and collapsing the two is what let a 100% +// frame-loss wiring skew look identical to a healthy relay refusing the odd +// cross-account edit. +// +// A contract defect is therefore its own bucket with its own counter: still +// non-fatal (every frame carries the same defect, so tearing the stream down +// would just kill the relay outright), but logged as "the relay is misconfigured +// and nothing is committing" rather than "one frame refused". +// +// The tests below pin BOTH halves of the distinction — that a defect increments +// contractDefects and NOT refusedFrames, and that an ordinary refusal still +// increments refusedFrames and NOT contractDefects. Without the negative half a +// bug that counted every drop in both buckets would pass. + +import ( + "context" + "errors" + "testing" + + "connectrpc.com/connect" + + compassv1internal "github.com/sealedsecurity/compass/go/internal/gen/compass/v1" +) + +// THE gap test. This drives the frame the production agent ACTUALLY emits today +// and asserts what genuinely happens: nothing commits. +// +// The first-party agent's only conversation-frame emitter is +// EventMapper.#appendBlock (packages/compass-agent/src/mapping.ts:386-393). It +// builds a Message from `blocks` alone and never stamps an id — the comment there +// states outright that the agent has no server id to mint — and nothing between +// the agent and this hub stamps one either (relay.go forwards the frame verbatim; +// the Runner's PostConversationFrame is unimplemented). No production path emits +// conversationPosted at all. So EVERY real agent turn arrives here as an id-less +// conversation_updated, and an update with no id addresses no row. +// +// This test therefore asserts the DEFECT, deliberately: zero sink calls, the +// contract-defect counter at 1, the refusal counter still 0, and the stream alive. +// It is the pin that keeps the gap a known, asserted state instead of a latent +// surprise — the write-through committing 0% of agent turns is currently +// indistinguishable, from the outside, from it committing 100%. +// +// WHEN RUNNER-SIDE ID RECONCILIATION LANDS (compass-runner lane, not this one), +// FLIP THIS TEST: the Runner will stamp the server id it learned from the first +// commit onto each subsequent update, the frame will address a real row, and the +// assertions below become "the sink saw one call and a row committed". Until +// then, a green run of this test means the gap is still open and still known. +func TestDeliverIDLessConversationUpdateIsAContractDefectAndCommitsNothing(t *testing.T) { + hub, conv, life, tail := newHub() + bindSession(hub, "sess-1") + + // Exactly the shape #appendBlock produces: blocks, no id. + err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "sess-1", + Frame: convUpdatedFrameIDLess("the agent speaks, and it is lost"), + }) + if err != nil { + t.Fatalf("Deliver(id-less update) = %v, want nil — a contract defect must not tear the stream down (every frame carries it, so a teardown would kill the relay outright)", err) + } + + // Nothing commits. The sink is never even reached: the hub can see the + // defect in the frame itself, so it refuses before attempting a write. + if got := len(conv.snapshot()); got != 0 { + t.Fatalf("conversation sink saw %d calls for an id-less update, want 0 — an update with no id addresses no row, so there is nothing to attempt", got) + } + if got := len(life.snapshot()) + len(tail.snapshot()); got != 0 { + t.Fatalf("an id-less conversation frame reached %d other sink calls, want 0", got) + } + + // It lands in the CONTRACT-DEFECT bucket... + if got := hub.ContractDefects(); got != 1 { + t.Fatalf("ContractDefects = %d, want 1 — an id-less relayed update is a wiring/contract skew between agent and Server, not routine per-frame garbage", got) + } + // ...and NOT the ordinary-refusal bucket. This is the load-bearing negative: + // if the defect were counted as a refusal, total frame loss would read as a + // healthy relay refusing frames and the operator would have no signal. + if got := hub.RefusedFrames(); got != 0 { + t.Fatalf("RefusedFrames = %d, want 0 — a contract defect must NOT be indistinguishable from an authz refusal", got) + } + + // The stream survives: the same hub carries the next frame. + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 2, + SessionID: "sess-1", + Frame: convPostedFrame("the next frame still flows"), + }); err != nil { + t.Fatalf("Deliver(following frame) = %v, want nil — the stream must survive a contract defect", err) + } + if got := len(conv.snapshot()); got != 1 { + t.Fatalf("conversation sink saw %d calls after the following frame, want 1", got) + } + if got := hub.ContractDefects(); got != 1 { + t.Fatalf("ContractDefects = %d after a good frame, want still 1", got) + } +} + +// Every real agent turn carries the same defect, so the counter must ACCUMULATE +// rather than latch — a rising count is the only way an operator distinguishes +// "the relay is dead" from "one frame was malformed once". A latching bool or a +// once-only log would make sustained total loss look like a single hiccup. +func TestDeliverIDLessConversationUpdatesAccumulateTheDefectCount(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + + for seq := uint64(1); seq <= 3; seq++ { + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: seq, + SessionID: "sess-1", + Frame: convUpdatedFrameIDLess("streaming turn"), + }); err != nil { + t.Fatalf("Deliver(id-less update, seq %d) = %v, want nil", seq, err) + } + } + + if got := hub.ContractDefects(); got != 3 { + t.Fatalf("ContractDefects = %d after three id-less updates, want 3 — the count must accumulate so sustained total loss is visible", got) + } + if got := len(conv.snapshot()); got != 0 { + t.Fatalf("conversation sink saw %d calls, want 0", got) + } + if got := hub.RefusedFrames(); got != 0 { + t.Fatalf("RefusedFrames = %d, want 0", got) + } +} + +// The other half of the distinction, from the error side: a sink error carrying +// CodeFailedPrecondition is a contract defect too (the comms layer's vocabulary +// for a structurally impossible request — e.g. a session bound to a NON-AGENT +// account, comms/agent_caller.go homeChannel), while NotFound/InvalidArgument +// stay ordinary refusals. Without this the hub would have only the frame-shape +// check and a bad binding would read as routine garbage. +func TestDeliverClassifiesSinkErrorsIntoRefusalVersusContractDefect(t *testing.T) { + for _, tc := range []struct { + name string + err error + wantRefused uint64 + wantDefects uint64 + wantStreamAlive bool + }{ + { + name: "cross-account update is an ordinary refusal", + err: connect.NewError(connect.CodeNotFound, errors.New("message not found")), + wantRefused: 1, + wantDefects: 0, + wantStreamAlive: true, + }, + { + name: "malformed block set is an ordinary refusal", + err: connect.NewError(connect.CodeInvalidArgument, errors.New("ask has no ask_id")), + wantRefused: 1, + wantDefects: 0, + wantStreamAlive: true, + }, + { + name: "non-agent binding is a contract defect", + err: connect.NewError(connect.CodeFailedPrecondition, errors.New("comms: account is not an agent account")), + wantRefused: 0, + wantDefects: 1, + wantStreamAlive: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + conv.err = tc.err + + err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "sess-1", + Frame: convPostedFrame("classify me"), + }) + if tc.wantStreamAlive && err != nil { + t.Fatalf("Deliver = %v, want nil (both classes are non-fatal drops)", err) + } + if got := hub.RefusedFrames(); got != tc.wantRefused { + t.Fatalf("RefusedFrames = %d, want %d", got, tc.wantRefused) + } + if got := hub.ContractDefects(); got != tc.wantDefects { + t.Fatalf("ContractDefects = %d, want %d", got, tc.wantDefects) + } + }) + } +} + +// THE fail-safe default, pinned. isFrameRefusal keys on the Connect code, so the +// drop-vs-teardown decision silently depends on every ConversationSink routing +// its errors through edgeError. A sink that returns a BARE error (an +// errors.New that never met the mapping) must therefore tear the stream down — +// connect.CodeOf reports CodeUnknown for it, which is neither a refusal nor a +// defect, so it falls through to the fault path. +// +// That is the right default (an unclassified error is not evidence the frame was +// bad), but it was only implicit in connect.CodeOf's behavior. This test makes +// it a contract: an unmapped sink error is FATAL and counts as neither a refusal +// nor a defect, so a future sink that forgets edgeError fails loudly at the +// stream rather than quietly shredding frames into a counter. +func TestDeliverBareSinkErrorTearsDownTheStreamAndCountsNothing(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + conv.err = errors.New("boom") + + err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "sess-1", + Frame: convPostedFrame("unclassified failure"), + }) + if err == nil { + t.Fatal("Deliver(bare sink error) = nil, want an error — an error that never met edgeError is unclassified, and an unclassified error must not be assumed to be the frame's fault") + } + if got := hub.RefusedFrames(); got != 0 { + t.Fatalf("RefusedFrames = %d, want 0 — an unmapped error is not a per-frame refusal", got) + } + if got := hub.ContractDefects(); got != 0 { + t.Fatalf("ContractDefects = %d, want 0 — an unmapped error is not a contract defect either", got) + } +} + +// The diagnostic surface: the three frame-loss counters plus the gap flag are +// readable as one snapshot, and the hub emits that snapshot on demand. Before +// this, total frame loss was visible ONLY in per-frame warn lines — there was no +// way to ask the hub "how much have you dropped", which is exactly the question +// a wedged relay raises. +func TestFrameDiagnosticsSnapshotReportsEveryLossCounter(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + + // One of each loss class: an unknown variant, an ordinary refusal, a + // contract defect, and a sequence gap. + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, SessionID: "sess-1", Frame: &compassv1internal.AgentFrame{}, + }); err != nil { + t.Fatalf("Deliver(unknown frame) = %v, want nil", err) + } + conv.err = connect.NewError(connect.CodeNotFound, errors.New("message not found")) + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 2, SessionID: "sess-1", Frame: convPostedFrame("refused"), + }); err != nil { + t.Fatalf("Deliver(refused frame) = %v, want nil", err) + } + conv.err = nil + // Seq 4 skips 3: an in-transit gap. + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 4, SessionID: "sess-1", Frame: convUpdatedFrameIDLess("id-less"), + }); err != nil { + t.Fatalf("Deliver(id-less update) = %v, want nil", err) + } + + got := hub.FrameDiagnostics() + want := FrameDiagnostics{SeenGap: true, UnknownFrames: 1, RefusedFrames: 1, ContractDefects: 1} + if got != want { + t.Fatalf("FrameDiagnostics() = %+v, want %+v", got, want) + } + // The snapshot agrees with the individual accessors — they read the same + // state under the same lock, so a drift here means one of them is stale. + if got.UnknownFrames != hub.UnknownFrames() || + got.RefusedFrames != hub.RefusedFrames() || + got.ContractDefects != hub.ContractDefects() || + got.SeenGap != hub.SeenGap() { + t.Fatalf("FrameDiagnostics() %+v disagrees with the individual accessors", got) + } +} diff --git a/go/internal/runnerhub/conversation_writethrough_test.go b/go/internal/runnerhub/conversation_writethrough_test.go new file mode 100644 index 00000000..83f25440 --- /dev/null +++ b/go/internal/runnerhub/conversation_writethrough_test.go @@ -0,0 +1,198 @@ +//go:build unix + +package runnerhub + +// Deliver's conversation arms after the T3 write-through landed (SEA-1364): the +// session->account resolution that precedes the sink, and the refusal posture +// that keeps a rejected frame from tearing down the Runner's PublishEvents +// stream. Two invariants, each with a plausible regression: +// +// - FAIL CLOSED. An unbound session must NOT reach the sink at all. Before +// T3 the sink was a log line, so an unresolved session cost nothing; now the +// sink COMMITS, and a frame that arrived with no resolvable account must +// never be attributed to a default, a stale binding, or the bootstrap admin. +// A regression that resolved to "" and let the call through would push the +// decision down to the comms layer's errNoActor guard — which would catch +// it, but one layer too late and only by luck. +// +// - REFUSALS ARE NON-FATAL. A frame the comms layer refuses (cross-account, +// revoked member, malformed ask, empty message id) is ONE bad frame, not a +// broken transport. It must be dropped and logged, and the stream must go on +// to commit the next valid frame — the same posture countUnknown takes for an +// unrecognized variant (hub.go). Only a genuine store/transaction failure +// tears the stream down, because that one is not frame-specific and retrying +// the relay is the right answer. + +import ( + "context" + "errors" + "testing" + + "connectrpc.com/connect" + + compassv1internal "github.com/sealedsecurity/compass/go/internal/gen/compass/v1" + "github.com/sealedsecurity/compass/go/internal/store" +) + +// THE fail-closed test. A conversation frame for a session the hub never bound +// (never provisioned, already stopped, or dropped by a Runner reconnect) reaches +// NO sink and is refused. The sink seeing zero calls is the load-bearing +// assertion: it proves the refusal happens at the resolution site, before any +// commit is attempted, so no default identity can be substituted downstream. +// +// Mutation: resolve to a zero AccountID and call the sink anyway → the sink +// records a call and this reddens. Mutation: fall back to any default account → +// same, plus the account assertion in the happy-path test would drift. +func TestDeliverConversationUnboundSessionFailsClosed(t *testing.T) { + for _, tc := range []struct { + name string + frame *compassv1internal.AgentFrame + }{ + {"posted", convPostedFrame("never committed")}, + {"updated", convUpdatedFrame("never committed")}, + } { + t.Run(tc.name, func(t *testing.T) { + hub, conv, life, tail := newHub() + // Deliberately NO bindSession: this id resolves to nothing. + + err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "never-bound", + Frame: tc.frame, + }) + if err != nil { + t.Fatalf("Deliver(unbound session) = %v, want nil (a refusal is a non-fatal drop, not a stream teardown)", err) + } + + if got := len(conv.snapshot()); got != 0 { + t.Fatalf("conversation sink saw %d calls for an unbound session, want 0 — an unresolvable frame must never reach the write-through", got) + } + if got := len(life.snapshot()) + len(tail.snapshot()); got != 0 { + t.Fatalf("an unbound conversation frame reached %d other sink calls, want 0", got) + } + if got := hub.RefusedFrames(); got != 1 { + t.Fatalf("RefusedFrames = %d, want 1 (a refusal is counted, never silently swallowed)", got) + } + }) + } +} + +// A frame whose session WAS bound but whose commit the comms layer refuses is +// also non-fatal: Deliver returns nil, the refusal is counted, and — the point +// of the test — the SAME hub then commits a following valid frame. Without the +// survival half, a bug that returned the refusal as a stream error would still +// pass a "returns nil" assertion if it only broke on the next frame. +// +// Each subtest names one refusal class the comms layer produces. They are +// indistinguishable to the hub (it sees a Connect error and drops), which is +// exactly why the hub tests them as a class rather than re-deriving comms' authz. +func TestDeliverConversationRefusalIsNonFatalAndStreamSurvives(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"cross-account update", connect.NewError(connect.CodeNotFound, errors.New("message not found"))}, + {"revoked member update", connect.NewError(connect.CodeNotFound, errors.New("message not found"))}, + {"malformed ask", connect.NewError(connect.CodeInvalidArgument, errors.New("ask has no ask_id"))}, + {"empty message id", connect.NewError(connect.CodeInvalidArgument, errors.New("message id is required"))}, + } { + t.Run(tc.name, func(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + conv.err = tc.err + + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "sess-1", + Frame: convUpdatedFrame("refused"), + }); err != nil { + t.Fatalf("Deliver(refused frame) = %v, want nil (one bad frame must not tear down the Runner stream)", err) + } + if got := hub.RefusedFrames(); got != 1 { + t.Fatalf("RefusedFrames = %d, want 1 (a refusal is observable, never silently swallowed)", got) + } + + // THE survival assertion: the stream goes on to commit the next + // valid frame on the same hub. + conv.err = nil + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 2, + SessionID: "sess-1", + Frame: convPostedFrame("the next good frame"), + }); err != nil { + t.Fatalf("Deliver(following valid frame) = %v, want nil — the stream must survive a refusal", err) + } + calls := conv.snapshot() + if len(calls) != 2 { + t.Fatalf("conversation sink saw %d calls, want 2 (the refused attempt and the following commit)", len(calls)) + } + if got := firstTextBlock(calls[1].posted.GetMessage()); got != "the next good frame" { + t.Fatalf("the frame after the refusal carried %q, want the following valid body", got) + } + if got := hub.RefusedFrames(); got != 1 { + t.Fatalf("RefusedFrames = %d after a successful commit, want still 1", got) + } + }) + } +} + +// The one error class that IS fatal: a genuine store/transaction failure is not +// frame-specific, so it ends the stream and the Runner retries the relay rather +// than losing the frame. This is the counterweight that keeps the non-fatal +// posture above from degenerating into "swallow everything" — a bug that +// dropped store faults too would make an outage look like a quiet stream. +func TestDeliverConversationStoreFailureTearsDownTheStream(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + conv.err = connect.NewError(connect.CodeInternal, errors.New("commit tx: connection reset")) + + err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, + SessionID: "sess-1", + Frame: convPostedFrame("lost to an outage"), + }) + if err == nil { + t.Fatal("Deliver(store failure) = nil, want an error (a store fault must end the stream so the Runner retries)") + } + if got := hub.RefusedFrames(); got != 0 { + t.Fatalf("RefusedFrames = %d, want 0 (a store fault is a teardown, not a per-frame refusal)", got) + } +} + +// The reconnect consequence, pinned deliberately rather than left to be +// rediscovered as a bug: enroll() clears ALL session bindings, so a frame +// in-flight across a Runner reconnect resolves to nothing and fails closed. That +// is the RULED behavior (a restarted Runner can re-mint a still-bound session id; +// clearing prevents attributing the new session's words to the old account). The +// fix is Runner-side resume, NOT a fallback here — a test that asserted the frame +// still committed after a reconnect would be asserting the security hole. +func TestDeliverConversationAfterRunnerReconnectFailsClosed(t *testing.T) { + hub, conv, _, _ := newHub() + bindSession(hub, "sess-1") + + // Positive control: while the binding is live the frame commits, so the + // refusal below is the reconnect and nothing else. + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 1, SessionID: "sess-1", Frame: convPostedFrame("before the reconnect"), + }); err != nil { + t.Fatalf("Deliver before the reconnect = %v, want nil", err) + } + if got := len(conv.snapshot()); got != 1 { + t.Fatalf("conversation sink saw %d calls before the reconnect, want 1", got) + } + + // The Runner reconnects: enroll drops every binding (OQ-2, ratified). + hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: 2, SessionID: "sess-1", Frame: convPostedFrame("after the reconnect"), + }); err != nil { + t.Fatalf("Deliver after the reconnect = %v, want nil (fail closed is a drop, not a teardown)", err) + } + if got := len(conv.snapshot()); got != 1 { + t.Fatalf("conversation sink saw %d calls after the reconnect, want still 1 — a re-minted session id must NOT inherit the old account", got) + } + if got := hub.RefusedFrames(); got != 1 { + t.Fatalf("RefusedFrames = %d, want 1", got) + } +} diff --git a/go/internal/runnerhub/helpers_test.go b/go/internal/runnerhub/helpers_test.go index 2931c7bf..521b228b 100644 --- a/go/internal/runnerhub/helpers_test.go +++ b/go/internal/runnerhub/helpers_test.go @@ -71,27 +71,32 @@ const testTimeout = 15 * time.Second func timeAfter() <-chan time.Time { return time.After(testTimeout) } -// convCall is one recorded PostAgentMessage: exactly one of posted/updated is -// non-nil, mirroring the ConversationSink contract. +// convCall is one recorded PostAgentMessage: the account the hub resolved the +// session to, the session id it resolved from, and exactly one of +// posted/updated — mirroring the ConversationSink contract. The account is the +// fake's proof of WHICH attribution the hub applied, the same security +// invariant the RelayCommsCall tests defend through commsCall. type convCall struct { + account store.AccountID sessionID string posted *compassv1.MessagePosted updated *compassv1.MessageUpdated } // fakeConversationSink records the conversation write-throughs Deliver drives so -// a test can assert which variant reached the comms surface, under which session -// id. Concurrency-safe: Deliver can run from a PublishEvents handler goroutine. +// a test can assert which variant reached the comms surface, under which account +// and session id. Concurrency-safe: Deliver can run from a PublishEvents handler +// goroutine. type fakeConversationSink struct { mu sync.Mutex calls []convCall err error // returned by PostAgentMessage when set (a write-through failure) } -func (f *fakeConversationSink) PostAgentMessage(_ context.Context, sessionID string, posted *compassv1.MessagePosted, updated *compassv1.MessageUpdated) error { +func (f *fakeConversationSink) PostAgentMessage(_ context.Context, account store.AccountID, sessionID string, posted *compassv1.MessagePosted, updated *compassv1.MessageUpdated) error { f.mu.Lock() defer f.mu.Unlock() - f.calls = append(f.calls, convCall{sessionID: sessionID, posted: posted, updated: updated}) + f.calls = append(f.calls, convCall{account: account, sessionID: sessionID, posted: posted, updated: updated}) return f.err } @@ -161,6 +166,25 @@ func newHubOnly() *Hub { return NewHub(&fakeConversationSink{}, &fakeLifecycleSink{}, &fakeTailSink{}, nil, discardLogger()) } +// testAgentAccount is the account every conversation test binds its session to. +// The tests assert WHICH account the hub attributed a frame to, so the value +// has to be nameable at the assertion site; it is a constant rather than a +// bindSession parameter because no test needs a second account — a cross +// -account case is built by binding this one and asserting a refusal, not by +// binding a different one. +const testAgentAccount store.AccountID = "acct-agent" + +// bindSession binds sessionID to testAgentAccount through the real +// Provision->Start promotion path, the same two-step the command handlers +// drive. Deliver's conversation arms resolve against this binding and fail +// closed without it, so any conversation test that expects a write-through must +// bind first. +func bindSession(hub *Hub, sessionID string) { + container := "container-for-" + sessionID + hub.bindContainer(container, testAgentAccount) + hub.promoteSession(container, sessionID) +} + // commsCall records one CommsCaller invocation: the account the hub resolved // the session to, and exactly one of the post/list request it dispatched. It is // the fake's proof of WHICH account attribution the hub applied — the security @@ -303,8 +327,32 @@ func convPostedFrame(text string) *compassv1internal.AgentFrame { } } -// convUpdatedFrame wraps a ConversationUpdated variant carrying one text block. +// convUpdatedFrame wraps a ConversationUpdated variant carrying one text block +// and an ADDRESSED message id — the shape an update must have to name the row it +// edits. No production emitter produces this yet (see convUpdatedFrameIDLess +// below); it is the shape the relay will carry once Runner-side id +// reconciliation lands, and the shape every test that expects a commit needs. func convUpdatedFrame(text string) *compassv1internal.AgentFrame { + return &compassv1internal.AgentFrame{ + Frame: &compassv1internal.AgentFrame_ConversationUpdated{ + ConversationUpdated: &compassv1.MessageUpdated{ + Message: &compassv1.Message{ + Id: "msg-addressed", + Blocks: []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: text}}}, + }, + }, + }, + } +} + +// convUpdatedFrameIDLess wraps a ConversationUpdated variant carrying one text +// block and NO message id — byte-for-byte the frame the first-party agent +// actually emits today (EventMapper.#appendBlock, +// packages/compass-agent/src/mapping.ts:386-393, which builds a Message from +// `blocks` alone because the agent has no server id to mint). Nothing between +// the agent and this hub stamps one, so this — not convUpdatedFrame — is the +// production shape on the current base. +func convUpdatedFrameIDLess(text string) *compassv1internal.AgentFrame { return &compassv1internal.AgentFrame{ Frame: &compassv1internal.AgentFrame_ConversationUpdated{ ConversationUpdated: &compassv1.MessageUpdated{ diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index 205b813a..e235966e 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -20,6 +20,8 @@ import ( "log/slog" "sync" + "connectrpc.com/connect" + compassv1 "github.com/sealedsecurity/compass/go/gen/compass/v1" compassv1internal "github.com/sealedsecurity/compass/go/internal/gen/compass/v1" "github.com/sealedsecurity/compass/go/internal/store" @@ -46,11 +48,42 @@ type RunnerEvent struct { // indistinguishable downstream from one a human posted. The comms package // implements it; the hub depends only on this narrow surface so it does not pull // the whole CommsService in. +// +// The sink takes the RESOLVED account, not just the session id: the hub owns the +// session->account binding and resolves it once at the Deliver site, exactly as +// RelayCommsCall does (relay_comms.go). Handing the sink a session id to resolve +// for itself would put the binding in two places, which is how attribution +// drifts — so the account is resolved here and passed, and the sink never looks +// a session up. type ConversationSink interface { // PostAgentMessage commits a conversation posted/updated event to the store - // and fans it out on the comms bus. posted distinguishes a new message - // (MessagePosted) from an update to one being composed (MessageUpdated). - PostAgentMessage(ctx context.Context, sessionID string, msg *compassv1.MessagePosted, updated *compassv1.MessageUpdated) error + // and fans it out on the comms bus, under account — the agent account the + // hub resolved the frame's session to. Exactly one of posted/updated is + // non-nil: posted is a new message (MessagePosted), updated an edit to one + // being composed (MessageUpdated). sessionID is carried for diagnostics + // only; account is the authority. + // + // ERROR VOCABULARY — REQUIRED OF EVERY IMPLEMENTATION. Deliver classifies a + // returned error by its CONNECT CODE alone (isFrameRefusal, + // isContractDefect), so an implementation MUST return connect errors whose + // codes carry the intended meaning — in practice by routing every store + // error through the comms package's edgeError mapping, exactly as a human + // caller's RPC does. The code decides the frame's fate: + // + // - NotFound / InvalidArgument → a per-frame refusal: the frame is + // dropped and counted, and the relay stream carries on. + // - FailedPrecondition → a contract defect: also dropped and counted, but + // against a separate counter and logged as a systemic misconfiguration, + // because a defect afflicts every frame rather than this one. + // - anything else, INCLUDING an unmapped bare error → an infrastructure + // fault: it ENDS the Runner's PublishEvents stream so the relay retries. + // + // That last clause is the fail-safe and it is deliberate: connect.CodeOf + // reports CodeUnknown for a plain errors.New, so a sink that forgets the + // mapping tears the stream down loudly instead of having its errors silently + // reinterpreted as "the frame was bad". Never return a bare error to mean a + // refusal — say it with a code. + PostAgentMessage(ctx context.Context, account store.AccountID, sessionID string, msg *compassv1.MessagePosted, updated *compassv1.MessageUpdated) error } // LifecycleSink publishes an extracted agent-session lifecycle transition onto @@ -124,6 +157,20 @@ type Hub struct { // unknownFrames counts frames whose oneof variant was unset or unrecognized // — logged and counted, never silently dropped (agent.proto:38-39). unknownFrames uint64 + // refusedFrames counts conversation frames the write-through refused as ONE + // bad frame — an unresolvable session, or a per-frame rejection from the + // comms layer (cross-account, revoked member, malformed frame). Each is a + // non-fatal drop: logged and counted, never silently swallowed, and never a + // stream teardown (that is reserved for store/transaction faults). + refusedFrames uint64 + // contractDefects counts conversation frames dropped because the agent and + // the Server disagree about the frame's SHAPE — a relayed update carrying no + // message id, or a session bound to a non-agent account. These are wiring + // skew, not per-frame garbage: unlike a refusal, EVERY frame carries the + // same defect, so a non-zero count means the relay is committing nothing at + // all. Counted separately from refusedFrames precisely so total loss cannot + // masquerade as a healthy relay refusing the occasional frame. + contractDefects uint64 } // attachedRunner is one enrolled Runner: its id, its authenticated token @@ -162,15 +209,19 @@ func NewHub(conversation ConversationSink, lifecycle LifecycleSink, tail Session // session-tail stream, plus its extracted lifecycle state → SubscribeEvents; a // frame whose variant is unset or unrecognized is logged and counted, never // silently dropped (design.md:1427-1434, agent.proto:38-39). +// +// A returned error ENDS the Runner's PublishEvents stream (handler.go:99-104), +// so only a fault the relay should retry is returned: a per-frame refusal is a +// non-fatal drop (deliverConversation), never a teardown. func (h *Hub) Deliver(ctx context.Context, ev RunnerEvent) error { h.recordSeq(ev.RunnerSeq) frame := ev.Frame switch f := frame.GetFrame().(type) { case *compassv1internal.AgentFrame_ConversationPosted: - return h.conversation.PostAgentMessage(ctx, ev.SessionID, f.ConversationPosted, nil) + return h.deliverConversation(ctx, ev, f.ConversationPosted, nil) case *compassv1internal.AgentFrame_ConversationUpdated: - return h.conversation.PostAgentMessage(ctx, ev.SessionID, nil, f.ConversationUpdated) + return h.deliverConversation(ctx, ev, nil, f.ConversationUpdated) case *compassv1internal.AgentFrame_Session: h.deliverSession(ev.SessionID, f.Session) return nil @@ -183,6 +234,39 @@ func (h *Hub) Deliver(ctx context.Context, ev RunnerEvent) error { } } +// isFrameRefusal reports whether err is a per-frame rejection rather than an +// infrastructure fault. The write-through surfaces comms/store errors through +// the same edgeError mapping a human caller gets, so the Connect code IS the +// classification: NotFound is the D9 collapse (an unknown message, a foreign +// one, or one whose channel membership was revoked) and InvalidArgument a +// malformed frame (empty block set, an ask missing its immutable id). Neither is +// retriable and neither implicates the transport. Everything else — notably +// CodeInternal from a failed transaction — is a real fault the relay should +// retry, so it is NOT swallowed. +func isFrameRefusal(err error) bool { + switch connect.CodeOf(err) { + case connect.CodeNotFound, connect.CodeInvalidArgument: + return true + default: + return false + } +} + +// isContractDefect reports whether err is a structural skew between the relay's +// ends rather than one bad frame: CodeFailedPrecondition is the vocabulary a +// sink uses for a request that is impossible as posed, not merely refused — a +// session bound to a NON-AGENT account (comms/agent_caller.go homeChannel), which +// no retry and no other frame can fix. +// +// The distinction matters because the two buckets answer different operator +// questions. A rising refusedFrames says "an authz boundary is being hit"; a +// rising contractDefects says "this relay is committing nothing, go fix the +// wiring". Collapsing them, as the pre-review code did, made those two states +// indistinguishable. +func isContractDefect(err error) bool { + return connect.CodeOf(err) == connect.CodeFailedPrecondition +} + // SeenGap reports whether Deliver ever observed a Runner-sequence gap. For the // board/diagnostics; a gap means in-transit loss the Client bus resync recovers. func (h *Hub) SeenGap() bool { @@ -198,6 +282,143 @@ func (h *Hub) UnknownFrames() uint64 { return h.unknownFrames } +// RefusedFrames reports the count of conversation frames the write-through +// refused — an unresolvable session, or a per-frame rejection from the comms +// layer. Non-zero is meaningful: a healthy relay refuses nothing, so a rising +// count is either a real authz boundary being hit or a wiring skew worth +// investigating (the diagnostic sibling of UnknownFrames). +func (h *Hub) RefusedFrames() uint64 { + h.mu.Lock() + defer h.mu.Unlock() + return h.refusedFrames +} + +// ContractDefects reports the count of conversation frames dropped because the +// relay's two ends disagree about the frame's shape — an id-less relayed update, +// or a session bound to a non-agent account. It reads differently from +// RefusedFrames on purpose: a refusal is one bad frame, a defect is a broken +// relay, and on the current base EVERY agent turn is a defect (the agent emits +// id-less updates and no hop stamps an id — see deliverConversation's shape +// check). So a defect count that tracks the frame count means nothing is +// committing at all. +func (h *Hub) ContractDefects() uint64 { + h.mu.Lock() + defer h.mu.Unlock() + return h.contractDefects +} + +// FrameDiagnostics is the hub's frame-loss snapshot: every way a relayed frame +// fails to reach its surface, plus the in-transit gap flag, read together under +// one lock so the four values describe the same instant. Reading the accessors +// one by one would interleave with Deliver and could report a gap without the +// drop that accompanied it. +type FrameDiagnostics struct { + // SeenGap is true once a Runner-sequence gap was observed (in-transit loss + // the Client bus resync recovers). + SeenGap bool + // UnknownFrames counts frames whose oneof variant was unset or unrecognized. + UnknownFrames uint64 + // RefusedFrames counts conversation frames refused as one bad frame. + RefusedFrames uint64 + // ContractDefects counts conversation frames dropped for relay wiring skew. + // Non-zero and rising is the loud one: the relay is committing nothing. + ContractDefects uint64 +} + +// FrameDiagnostics returns the frame-loss snapshot. It exists because the three +// counters had no non-test reader: total frame loss was observable only by +// reading warn logs, which is no way to answer "is this relay committing +// anything". serve.go logs this snapshot on shutdown (sinks.go, +// LogFrameDiagnostics), so a run that dropped every agent turn says so once, in +// one line, rather than only in the per-frame noise. +func (h *Hub) FrameDiagnostics() FrameDiagnostics { + h.mu.Lock() + defer h.mu.Unlock() + return FrameDiagnostics{ + SeenGap: h.seenGap, + UnknownFrames: h.unknownFrames, + RefusedFrames: h.refusedFrames, + ContractDefects: h.contractDefects, + } +} + +// deliverConversation resolves the frame's session to its agent account and +// write-throughs the frame under that account, classifying any failure. +// +// Resolution is the SAME binding RelayCommsCall resolves against +// (accountForSession, relay_comms.go:78) and it happens HERE, once, rather than +// inside the sink: the hub owns the session->account binding, and a second +// resolution site is how attribution drifts. +// +// Fail closed on an unbound session. The account is the agent's IDENTITY on a +// durable, fanned-out row — there is no safe default, so an unresolvable session +// is refused outright rather than committed under a fallback. This mirrors +// errNoActor (comms/agent_caller.go): a missing actor is a hard refusal +// precisely so a wiring bug cannot quietly attribute an agent's words to the +// bootstrap admin. The refusal is a DROP, not a teardown — one frame the Server +// cannot attribute must not kill the Runner's whole event stream. +// +// KNOWN CONSEQUENCE, deliberate — do not "fix" it with a fallback. enroll() +// (:269) calls clear(h.sessionAccounts) on every Runner re-enroll, because +// session ids are Runner-minted and a restarted Runner can re-mint an id still +// bound here; clearing is what stops the new session's words being attributed to +// the previous account. So a frame in flight across a Runner reconnect resolves +// to nothing and lands on this refusal path. That is the ruled behavior (OQ-2, +// ratified). Reintroducing any fallback here — a default account, a retained +// stale binding — would reopen exactly the misattribution the clear() exists to +// close. The right fix is Runner-side session resume, which is another lane's +// work; it re-binds the session and this path then resolves normally. +func (h *Hub) deliverConversation( + ctx context.Context, + ev RunnerEvent, + posted *compassv1.MessagePosted, + updated *compassv1.MessageUpdated, +) error { + account, ok := h.accountForSession(ev.SessionID) + if !ok { + h.countRefused(ev, "no agent account bound to the frame's session") + return nil + } + // Frame-shape check BEFORE the sink. An update names the row it edits by + // message.id, so an id-less one addresses nothing and there is no write to + // attempt — the comms layer would reject it as InvalidArgument, which is + // indistinguishable here from an ordinary malformed frame. Checking the shape + // at the seam is what lets the hub name the real cause, and this is the + // SOLE frame shape the production agent emits today (see contractDefect's + // doc and contract_defect_test.go), so misclassifying it would report 100% + // frame loss as routine refusals. + if updated != nil && updated.GetMessage().GetId() == "" { + h.countContractDefect(ev, "relayed conversation_updated carries no message.id, so it addresses no row and nothing can commit") + return nil + } + if err := h.conversation.PostAgentMessage(ctx, account, ev.SessionID, posted, updated); err != nil { + switch { + case isContractDefect(err): + // Not this frame's fault: the relay is wired wrong (a session bound + // to a non-agent account, a Server-side dispatch skew). Still a + // drop, because every frame carries the same defect and a teardown + // would kill the relay outright — but counted and logged as the + // systemic fault it is. + h.countContractDefect(ev, err.Error()) + return nil + case isFrameRefusal(err): + // One frame the comms layer rejected — a cross-account or revoked + // -member edit, or a malformed frame. It is specific to this frame, + // so retrying the relay would only replay the same rejection: drop + // and count it, and let the stream carry the next frame. + h.countRefused(ev, err.Error()) + return nil + } + // Not frame-specific (a store or transaction fault), or unclassified + // (an error that never met edgeError, which connect.CodeOf reports as + // CodeUnknown): end the stream so the Runner retries the relay rather + // than losing the frame. An error the Server cannot classify is never + // assumed to be the frame's fault. + return err + } + return nil +} + // deliverSession routes a session frame to the observation-pane tail and, when // the frame carries a lifecycle transition, extracts the AgentSessionStatus onto // SubscribeEvents. A session frame can carry a trace event, a lifecycle @@ -240,6 +461,51 @@ func (h *Hub) countUnknown(ev RunnerEvent) { slog.Uint64("unknown_total", n)) } +// countRefused records and logs a refused conversation frame — the write-through +// counterpart of countUnknown, and the same posture: the frame is dropped, but +// the drop is COUNTED and LOGGED so it is observable. A refusal is never silent, +// because silence here would hide both a real authz boundary being hit and a +// binding/wiring skew that is losing an agent's words. +func (h *Hub) countRefused(ev RunnerEvent, reason string) { + h.mu.Lock() + h.refusedFrames++ + n := h.refusedFrames + h.mu.Unlock() + h.log.Warn("refused relayed conversation frame (dropped, stream continues)", + slog.String("session_id", ev.SessionID), + slog.Uint64("runner_seq", ev.RunnerSeq), + slog.String("reason", reason), + slog.Uint64("refused_total", n)) +} + +// countContractDefect records and logs a conversation frame dropped for relay +// wiring skew. Same non-fatal posture as countRefused — the frame is dropped and +// the stream lives — but a deliberately louder line, because the two say very +// different things to whoever reads them. +// +// countRefused's line means "one frame was rejected"; a reader can reasonably +// ignore it. THIS line means the relay's two ends disagree structurally, so every +// frame is being lost and no retry, redeploy, or next frame will change that. +// The message therefore states the consequence outright rather than naming the +// frame: an operator scanning logs must not have to infer total data loss from a +// counter they were never told to watch. +// +// Error, not Warn: a refusal is expected traffic on a healthy system, a contract +// defect never is. It stays non-fatal all the same — returning an error here +// would end the Runner's PublishEvents stream on the FIRST agent turn, since on +// the current base every turn carries the defect. +func (h *Hub) countContractDefect(ev RunnerEvent, reason string) { + h.mu.Lock() + h.contractDefects++ + n := h.contractDefects + h.mu.Unlock() + h.log.Error("agent conversation relay is misconfigured: NOTHING is being committed to comms", + slog.String("session_id", ev.SessionID), + slog.Uint64("runner_seq", ev.RunnerSeq), + slog.String("defect", reason), + slog.Uint64("contract_defect_total", n)) +} + // enroll registers (or re-attaches) a Runner under its authenticated subject, // returning whether it re-attached an existing Runner (OQ6 duplicate enrollment: // a second enrollment re-attaches the same Runner rather than registering a diff --git a/go/internal/runnerhub/hub_test.go b/go/internal/runnerhub/hub_test.go index 6ee88b44..02abcd09 100644 --- a/go/internal/runnerhub/hub_test.go +++ b/go/internal/runnerhub/hub_test.go @@ -80,9 +80,12 @@ func TestDeliverSessionTraceOnlyPublishesNoLifecycle(t *testing.T) { // Routing: a conversation-posted frame hits ONLY the ConversationSink (as a // posted message), and NOT the tail or lifecycle. Mis-routing a conversation to -// the tail would silently drop it from the comms store of record. +// the tail would silently drop it from the comms store of record. The frame +// reaches the sink under the account the hub resolved the session to — the +// attribution the write-through commits under. func TestDeliverConversationPostedRoutesToCommsOnly(t *testing.T) { hub, conv, life, tail := newHub() + bindSession(hub, "sess-conv") if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, @@ -97,6 +100,9 @@ func TestDeliverConversationPostedRoutesToCommsOnly(t *testing.T) { t.Fatalf("conversation sink saw %d calls, want 1", len(calls)) } c := calls[0] + if c.account != "acct-agent" { + t.Fatalf("conversation account = %q, want the bound acct-agent (the hub resolves session->account before the sink)", c.account) + } if c.sessionID != "sess-conv" { t.Fatalf("conversation session id = %q, want sess-conv", c.sessionID) } @@ -119,10 +125,11 @@ func TestDeliverConversationPostedRoutesToCommsOnly(t *testing.T) { } // A conversation-updated frame arrives on the ConversationSink as the UPDATED -// arg (posted nil) — the streaming-turn path. A bug that swapped the posted and -// updated args would redden the nil checks. +// arg (posted nil) — the streaming-turn path — under the same resolved account. +// A bug that swapped the posted and updated args would redden the nil checks. func TestDeliverConversationUpdatedRoutesAsUpdated(t *testing.T) { hub, conv, _, _ := newHub() + bindSession(hub, "sess-conv") if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, @@ -136,6 +143,9 @@ func TestDeliverConversationUpdatedRoutesAsUpdated(t *testing.T) { if len(calls) != 1 { t.Fatalf("conversation sink saw %d calls, want 1", len(calls)) } + if calls[0].account != "acct-agent" { + t.Fatalf("conversation account = %q, want the bound acct-agent", calls[0].account) + } if calls[0].updated == nil { t.Fatalf("updated arg was nil; an updated frame must arrive as the updated arg") } diff --git a/go/internal/runnerhub/integration_pgtest_test.go b/go/internal/runnerhub/integration_pgtest_test.go index ec34e1d2..88ae83f2 100644 --- a/go/internal/runnerhub/integration_pgtest_test.go +++ b/go/internal/runnerhub/integration_pgtest_test.go @@ -338,7 +338,7 @@ func openStoreFixture(t *testing.T, ctx context.Context, dsn string) (*store.Sto // CommsCaller leg handles instead. type noopConversationSink struct{} -func (noopConversationSink) PostAgentMessage(context.Context, string, *compassv1.MessagePosted, *compassv1.MessageUpdated) error { +func (noopConversationSink) PostAgentMessage(context.Context, store.AccountID, string, *compassv1.MessagePosted, *compassv1.MessageUpdated) error { return nil } diff --git a/go/internal/runnerhub/seam_test.go b/go/internal/runnerhub/seam_test.go index e0c8005f..c8fc75bd 100644 --- a/go/internal/runnerhub/seam_test.go +++ b/go/internal/runnerhub/seam_test.go @@ -90,6 +90,14 @@ func TestSeamEnrollSessionsRoundTripAndPublishEvents(t *testing.T) { t.Fatalf("runner sessions loop ended with %v, want clean EOF", err) } + // Bind the relayed session to an agent account. Deliver's conversation arms + // resolve session->account and fail closed on an unbound session (hub.go, + // deliverConversation), so without a binding the frame below would be a + // counted refusal rather than reaching the sink. Production binds this at + // Provision->Start; this seam test drives PublishEvents directly, so it + // binds through the same promotion path the command handlers use. + bindSession(hub, "sess-wire") + // --- PublishEvents: one relayed frame must reach Hub.Deliver (a fake sink // observes it). This is the event path terminating the wire. ----------------- pub := client.PublishEvents(ctx) diff --git a/go/internal/store/messages.go b/go/internal/store/messages.go index 0913b69a..2b369d88 100644 --- a/go/internal/store/messages.go +++ b/go/internal/store/messages.go @@ -178,6 +178,96 @@ func updateMessageBlocksExec(ctx context.Context, db execer, id MessageID, block return nil } +// UpdateMessageBlocksAsAuthor replaces a message's block set UNDER an acting +// account — the only update path safe for a message id that arrives from +// outside the Server's own trust boundary (a relayed agent MessageUpdated +// frame, SEA-1364 T3). +// +// Why it is a fork and not a flag on the shared core. updateMessageBlocksExec +// addresses the row by a bare MessageID with NO membership and NO authorship +// check. That is correct where it is used — AnswerAsk has already resolved the +// target through a membership JOIN and locked it FOR UPDATE, so re-checking +// would be redundant, and AnswerAsk deliberately permits a MEMBER who is not +// the author to answer. Folding this path onto that core would either strip the +// authz a relayed id requires or break every ask answered by anyone but the +// asker. The two predicates genuinely differ, so they are two statements. +// +// The predicate is membership AND authorship: the actor must be a member of the +// message's channel and be its author. Both halves are load-bearing — authorship +// alone would let an account edit its own past message in a channel it has since +// been removed from, and membership alone would let any member rewrite another +// account's words. A failure of EITHER half, and an id that names no row at all, +// return the same ErrNotFound: the D9 not-found/forbidden merge, so an actor +// cannot learn that a message it may not touch exists (the same collapse +// AnswerAsk documents at :307-310). +// +// The updated row is returned via RETURNING, so the caller fans out the +// post-update state without a second read that could observe a later write. +// Validation mirrors updateMessageBlocksExec (an empty block set is rejected; an +// ask block must carry its immutable ask_id, since a re-minted id would orphan a +// pending RespondToAsk) and adds one this path needs: an EMPTY message id is +// ErrInvalidArgument rather than a zero-rows ErrNotFound, because a relayed +// MessageUpdated whose message.id was never stamped is a malformed frame, not an +// edit aimed at a row the actor cannot see. +// +// Deliberately NOT built on RequireAgentSessionSubscriber (agent_sessions.go:91). +// That primitive resolves the whole session-ownership chain in a SINGLE EXISTS +// query returning only ErrNotFound-or-nil, so an unknown session and a +// known-but-foreign one are indistinguishable by error class AND by timing — a +// deliberate anti-enumeration property. Reusing its joins here would mean +// splitting it into a resolve-then-check pair, which would look like a clean +// refactor, pass every test, and reintroduce a session-enumeration oracle. This +// path needs a message-scoped predicate, not a session-scoped one, so it is its +// own single-statement gate and that primitive is left completely alone. +func (s *Store) UpdateMessageBlocksAsAuthor(ctx context.Context, actor AccountID, id MessageID, blocks []MessageBlock) (Message, error) { + if id == "" { + return Message{}, fmt.Errorf("%w: message id is required", ErrInvalidArgument) + } + if len(blocks) == 0 { + return Message{}, fmt.Errorf("%w: message has no blocks", ErrInvalidArgument) + } + for i, b := range blocks { + if b.Ask != nil && b.Ask.AskID == "" { + return Message{}, fmt.Errorf("%w: block %d ask has no ask_id (an update must carry the existing id)", ErrInvalidArgument, i) + } + } + blocksJSON, err := marshalBlocks(blocks) + if err != nil { + return Message{}, err + } + + // One statement, so the authz predicate and the write cannot race: a + // membership revoked concurrently either lands before the UPDATE (which then + // matches no row) or after it, never between a separate check and the write. + // The EXISTS subquery is the membership half and the author_account_id + // equality the authorship half; both must hold for the row to match. + const q = ` + UPDATE messages m + SET blocks = $1, text_content = $2 + WHERE m.id = $3 + AND m.author_account_id = $4 + AND EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = m.channel_id AND cm.account_id = $4 + ) + RETURNING m.id, m.channel_id, m.author_account_id, m.at_unix_ms, m.blocks, COALESCE(m.parent_message_id, '')` + rows, err := s.pool.Query(ctx, q, blocksJSON, textContent(blocks), string(id), string(actor)) + if err != nil { + return Message{}, fmt.Errorf("store: update message blocks as author: %w", err) + } + defer rows.Close() + msgs, err := scanMessages(rows) + if err != nil { + return Message{}, err + } + if len(msgs) == 0 { + // Unknown id, not the author, or no longer a member — one answer for all + // three, so a refusal enumerates nothing. + return Message{}, fmt.Errorf("%w: message %q", ErrNotFound, id) + } + return msgs[0], nil +} + // ListMessages pages a channel's messages newest-first, clamped to the store's // page bounds (comms.proto:446-461). Ordering keys on the monotonic seq (a // stable total order even under equal timestamps); BeforeMessageID pages diff --git a/go/internal/store/messages_authored_update_test.go b/go/internal/store/messages_authored_update_test.go new file mode 100644 index 00000000..5d868189 --- /dev/null +++ b/go/internal/store/messages_authored_update_test.go @@ -0,0 +1,253 @@ +//go:build pgtest + +package store + +// UpdateMessageBlocksAsAuthor: the AUTHORIZING block-update path, the store leg +// of the relayed-agent-input write-through (SEA-1364 T3). Its whole reason to +// exist is that updateMessageBlocksExec takes a bare MessageID and +// performs NO membership or authorship check — safe for AnswerAsk, which has +// already gated on the actor's visible set, but a privilege hole on a path whose +// message id arrives from a relayed agent frame. Every test here defends one +// half of the predicate or one validation: +// +// - membership AND authorship both required, and both collapse to ErrNotFound +// (the D9 not-found/forbidden merge — an actor must not learn that a message +// it cannot touch exists); +// - the updated row comes back via RETURNING, so the caller can fan out the +// post-update state without a second read; +// - the validations updateMessageBlocksExec already enforces (empty block set, +// an ask block with an empty AskID) are enforced here too, plus an empty +// message id; +// - and, critically, AnswerAsk is untouched — its locked read-modify-write +// still runs through the un-authorized shared core. + +import ( + "context" + "reflect" + "testing" +) + +// authoredMessage appends one text message by author into ch and returns it — +// the row the authorizing-update cases then try to edit under various actors. +func authoredMessage(t *testing.T, ctx context.Context, s *Store, author AccountID, ch ChannelID, text string) Message { + t.Helper() + msg, _, err := s.AppendMessage(ctx, Message{ + Container: ContainerRef{ChannelID: ch}, AuthorAccountID: author, + Blocks: []MessageBlock{textBlock(text)}, + }, "") + if err != nil { + t.Fatalf("AppendMessage: %v", err) + } + return msg +} + +// The happy path: the author, who is a member of the message's channel, edits +// its blocks in place and gets the UPDATED row back from RETURNING — same id, +// same channel, same author, new blocks — and the edit is durable on re-read. +// +// Mutation that reddens it: returning the pre-update row (RETURNING on the old +// tuple), or dropping the text_content refresh (the re-read block assertion +// stays green but search would silently drift — covered by the read-back here +// only for blocks; the text index is asserted by the search suite). +func TestUpdateMessageBlocksAsAuthorEditsInPlaceAndReturnsRow(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + author := mustUser(t, s, "author") + ch := mustChannel(t, s, author.ID) + msg := authoredMessage(t, ctx, s, author.ID, ch.ID, "first draft") + + replacement := []MessageBlock{textBlock("settled turn"), askBlockID("ask-existing-7")} + updated, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, replacement) + if err != nil { + t.Fatalf("UpdateMessageBlocksAsAuthor: %v", err) + } + if updated.ID != msg.ID { + t.Fatalf("returned message id = %q, want the edited row %q", updated.ID, msg.ID) + } + if updated.Container.ChannelID != ch.ID { + t.Fatalf("returned channel = %q, want %q", updated.Container.ChannelID, ch.ID) + } + if updated.AuthorAccountID != author.ID { + t.Fatalf("returned author = %q, want %q", updated.AuthorAccountID, author.ID) + } + assertBlocksEqual(t, updated.Blocks, replacement) + + // Durable: a re-read through the normal visibility path sees the new set. + got, err := s.ListMessages(ctx, author.ID, ContainerRef{ChannelID: ch.ID}, Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(got) != 1 { + t.Fatalf("channel holds %d messages, want 1 (an update must edit in place, never insert)", len(got)) + } + assertBlocksEqual(t, got[0].Blocks, replacement) +} + +// THE cross-account security case. A different account — a full member of the +// same channel, so ONLY authorship separates it from the author — cannot edit +// the message, and gets ErrNotFound rather than a distinct forbidden. The +// positive control (the author succeeding on the same row) is what stops this +// passing vacuously if the update were broken for everyone. +// +// Mutation: drop `AND author_account_id = $actor` from the predicate → the +// member's edit succeeds and both the error assertion and the untouched-content +// assertion redden. +func TestUpdateMessageBlocksAsAuthorCrossAccountIsNotFound(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + author := mustUser(t, s, "author") + other := mustUser(t, s, "other") + // `other` is a founding member of the channel: it can READ the message, so + // the only thing refusing its edit is the authorship half of the predicate. + ch, err := s.CreateChannel(ctx, author.ID, NewChannel{ + Name: "room", Kind: ChannelKindChannel, MemberAccountIDs: []AccountID{other.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + msg := authoredMessage(t, ctx, s, author.ID, ch.ID, "the author's words") + + _, err = s.UpdateMessageBlocksAsAuthor(ctx, other.ID, msg.ID, []MessageBlock{textBlock("put into their mouth")}) + sentinelIs(t, err, ErrNotFound, "a channel member editing another account's message") + + // Positive control: the author CAN edit the same row, so the refusal above + // is authorship, not a blanket failure. + if _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, []MessageBlock{textBlock("the author's own edit")}); err != nil { + t.Fatalf("the author cannot edit its own message: %v", err) + } + + // The rejected edit left no trace: the row carries the author's own text. + got, err := s.ListMessages(ctx, author.ID, ContainerRef{ChannelID: ch.ID}, Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + assertBlocksEqual(t, got[0].Blocks, []MessageBlock{textBlock("the author's own edit")}) +} + +// The membership half: an account that AUTHORED the message but has since been +// removed from its channel cannot edit it — a revoked member is refused with the +// same ErrNotFound, so losing read access loses write access atomically. This is +// the case a bare authorship check would miss. +// +// Mutation: drop the channel_members EXISTS clause → the revoked author's edit +// succeeds and this reddens twice (error and content). +func TestUpdateMessageBlocksAsAuthorRevokedMemberIsNotFound(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + owner := mustUser(t, s, "owner") + author := mustUser(t, s, "author") + ch, err := s.CreateChannel(ctx, owner.ID, NewChannel{ + Name: "room", Kind: ChannelKindChannel, MemberAccountIDs: []AccountID{author.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + msg := authoredMessage(t, ctx, s, author.ID, ch.ID, "posted while a member") + + // Positive control BEFORE the revoke: while still a member the author can + // edit, so the post-revoke refusal is the membership change and nothing else. + if _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, []MessageBlock{textBlock("edited while a member")}); err != nil { + t.Fatalf("the author cannot edit while still a member: %v", err) + } + + if _, _, err := s.UpdateChannelMembers(ctx, owner.ID, ch.ID, []MemberUpdate{{AccountID: author.ID, Remove: true}}); err != nil { + t.Fatalf("UpdateChannelMembers(remove): %v", err) + } + + _, err = s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, []MessageBlock{textBlock("edited after the revoke")}) + sentinelIs(t, err, ErrNotFound, "a revoked member editing its own past message") + + // The revoked edit left no trace: the owner still reads the pre-revoke text. + got, err := s.ListMessages(ctx, owner.ID, ContainerRef{ChannelID: ch.ID}, Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + assertBlocksEqual(t, got[0].Blocks, []MessageBlock{textBlock("edited while a member")}) +} + +// An unknown message id is the same ErrNotFound as a message the actor may not +// touch — the two are indistinguishable, which is what makes the refusals above +// non-enumerating. +func TestUpdateMessageBlocksAsAuthorUnknownMessageIsNotFound(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + author := mustUser(t, s, "author") + _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, MessageID("ghost"), []MessageBlock{textBlock("x")}) + sentinelIs(t, err, ErrNotFound, "unknown message id") +} + +// The three input validations, each ErrInvalidArgument and each leaving the row +// untouched. Empty message id is new here (updateMessageBlocksExec lets it fall +// through to a zero-rows ErrNotFound); the empty block set and the empty AskID +// mirror the checks the shared core already makes, restated because this path +// deliberately does not call it. +func TestUpdateMessageBlocksAsAuthorRejectsMalformedInput(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + author := mustUser(t, s, "author") + ch := mustChannel(t, s, author.ID) + msg := authoredMessage(t, ctx, s, author.ID, ch.ID, "untouched") + + t.Run("empty message id", func(t *testing.T) { + // An empty id is malformed input, NOT a missing row: a relayed + // MessageUpdated whose message.id never got stamped is an agent/wire + // defect the caller should see as such, distinct from an edit aimed at a + // row the actor may not touch. + _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, MessageID(""), []MessageBlock{textBlock("x")}) + sentinelIs(t, err, ErrInvalidArgument, "empty message id") + }) + t.Run("empty block set", func(t *testing.T) { + _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, nil) + sentinelIs(t, err, ErrInvalidArgument, "empty block set") + }) + t.Run("ask block with no ask id", func(t *testing.T) { + // ask_id is minted once at append and immutable; an update carrying an + // id-less ask would orphan any pending RespondToAsk against the original. + _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, []MessageBlock{textBlock("x"), askBlockID("")}) + sentinelIs(t, err, ErrInvalidArgument, "ask block with an empty ask_id") + }) + + // None of the three wrote anything. + got, err := s.ListMessages(ctx, author.ID, ContainerRef{ChannelID: ch.ID}, Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + assertBlocksEqual(t, got[0].Blocks, []MessageBlock{textBlock("untouched")}) +} + +// The regression a shared-helper change would cause: AnswerAsk must keep working +// through the UN-authorized updateMessageBlocksExec core. AnswerAsk gates on +// membership itself (its FOR UPDATE find-and-lock JOINs channel_members) and +// deliberately allows a MEMBER who is not the author to answer — so folding it +// onto the authorship-requiring path would silently break every ask answered by +// anyone but the asker. This test answers as a non-author member precisely to +// pin that. +func TestAnswerAskStillWorksForANonAuthorMember(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + asker := mustUser(t, s, "asker") + answerer := mustUser(t, s, "answerer") + ch, err := s.CreateChannel(ctx, asker.ID, NewChannel{ + Name: "room", Kind: ChannelKindChannel, MemberAccountIDs: []AccountID{answerer.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + if _, _, err := s.AppendMessage(ctx, Message{ + Container: ContainerRef{ChannelID: ch.ID}, AuthorAccountID: asker.ID, + Blocks: []MessageBlock{textBlock("choose one"), pendingAsk("ask-1", false)}, + }, ""); err != nil { + t.Fatalf("AppendMessage: %v", err) + } + + updated, err := s.AnswerAsk(ctx, answerer.ID, "ask-1", []AskAnswer{{QuestionID: "q1", ChosenOptionIDs: []string{"opt-a"}}}) + if err != nil { + t.Fatalf("AnswerAsk by a non-author member: %v", err) + } + if got := updated.Blocks[1].Ask.Questions[0].ChosenOptionIDs; !reflect.DeepEqual(got, []string{"opt-a"}) { + t.Fatalf("returned chosen = %v, want [opt-a]", got) + } + if got := answeredAsk(t, ctx, s, asker.ID, ch.ID, "ask-1"); !reflect.DeepEqual(got, []string{"opt-a"}) { + t.Fatalf("persisted chosen = %v, want [opt-a]", got) + } +} diff --git a/go/server/serve.go b/go/server/serve.go index 1d0704f0..531de53e 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -16,6 +16,7 @@ import ( "context" "errors" "fmt" + "log/slog" "net" "net/http" "net/netip" @@ -33,6 +34,7 @@ import ( "github.com/sealedsecurity/compass/go/internal/auth" "github.com/sealedsecurity/compass/go/internal/board" "github.com/sealedsecurity/compass/go/internal/comms" + "github.com/sealedsecurity/compass/go/internal/runnerhub" "github.com/sealedsecurity/compass/go/internal/store" ) @@ -212,7 +214,10 @@ func Serve(ctx context.Context, cfg ServeConfig) error { // service's SubscribeAgentSession source (reader) — a frame the hub relays // fans to that session's stream subscribers. tail := newSessionTail() - hub := newRunnerHub(brd, tail, commsSvc, nil) + // One logger for the hub's per-frame diagnostics and for the frame-loss + // summary the drain logs, so both land on the same sink. + hubLog := slog.Default() + hub := newRunnerHub(brd, tail, commsSvc, hubLog) svc := newService(cfg.Version, bus, st, hub, brd, tail) // Shipped door: the Unix socket serves native gRPC (cleartext HTTP/2), @@ -300,44 +305,78 @@ func Serve(ctx context.Context, cfg ServeConfig) error { } // Drain member of the same group: wake on gctx cancellation (parent shutdown - // or a server erroring), close both buses first so held-open SubscribeEvents / - // SubscribeComms streams end and release their handlers, then drain the HTTP - // servers under a bounded deadline. A drain that overruns (a handler still - // wedged — e.g. a stream stuck mid replay to a stalled client) surfaces as the - // error rather than being swallowed into a false clean shutdown; because the - // group keeps the first error, a real serve error still wins over this drain. + // or a server erroring), then hand off to drainDoors. A drain that overruns + // (a handler still wedged — e.g. a stream stuck mid replay to a stalled + // client) surfaces as the error rather than being swallowed into a false + // clean shutdown; because the group keeps the first error, a real serve error + // still wins over this drain. g.Go(func() error { <-gctx.Done() - // Close both buses so held-open streams end and release their handlers: - // the CompassService SubscribeEvents rides bus, the CommsService - // SubscribeComms rides commsBus, and both serve on every door. Closing only - // bus would leave a live SubscribeComms subscriber wedged (its ctx is not - // cancelled by Shutdown), stalling the drain to the deadline. Both closes - // are idempotent, so the deferred Close of each stays a safe no-op. - bus.Close() - commsBus.Close() - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - drainErr := udsServer.Shutdown(shutdownCtx) //nolint:contextcheck // fresh ctx is deliberate: the parent is already cancelled (that woke this drain), so the bounded drain needs a live ctx, not the dead inherited one - if devServer != nil { - if err := devServer.Shutdown(shutdownCtx); err != nil && drainErr == nil { //nolint:contextcheck // same fresh-ctx rationale as the uds drain above - drainErr = err - } - } - if netServer != nil { - if err := netServer.Shutdown(shutdownCtx); err != nil && drainErr == nil { //nolint:contextcheck // same fresh-ctx rationale as the uds drain above - drainErr = err - } - } - if drainErr != nil { - return fmt.Errorf("draining compass.v1 servers on shutdown: %w", drainErr) - } - return nil + return drainDoors(drainSet{ //nolint:contextcheck // drainDoors deliberately takes no ctx: the inherited one is already cancelled (that cancellation is what woke this drain), so a bounded shutdown needs the fresh ctx drainDoors makes internally, not the dead one + bus: bus, + commsBus: commsBus, + uds: udsServer, + dev: devServer, + net: netServer, + hub: hub, + log: hubLog, + }) }) return g.Wait() } +// drainSet is the shutdown-side view of what Serve built: the two buses whose +// held-open streams must end before a door can drain, the doors themselves +// (dev and net are nil when not configured), and the hub whose frame-loss +// counters are final once every door is drained. +type drainSet struct { + bus *events.Bus[busPayload] + commsBus *events.Bus[*compassv1.SubscribeCommsResponse] + uds *http.Server + dev *http.Server + net *http.Server + hub *runnerhub.Hub + log *slog.Logger +} + +// drainDoors ends the live streams, drains every configured door under one +// bounded deadline, and reports the hub's final frame accounting. It is split +// out of Serve so the shutdown sequence reads as one unit: the ordering here is +// load-bearing and was previously buried at the bottom of a 200-line function. +func drainDoors(d drainSet) error { + // Close both buses so held-open streams end and release their handlers: + // the CompassService SubscribeEvents rides bus, the CommsService + // SubscribeComms rides commsBus, and both serve on every door. Closing only + // bus would leave a live SubscribeComms subscriber wedged (its ctx is not + // cancelled by Shutdown), stalling the drain to the deadline. Both closes + // are idempotent, so the deferred Close of each stays a safe no-op. + d.bus.Close() + d.commsBus.Close() + // Fresh ctx is deliberate: the parent is already cancelled (that woke this + // drain), so the bounded drain needs a live ctx, not the dead inherited one. + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + drainErr := d.uds.Shutdown(shutdownCtx) + for _, srv := range []*http.Server{d.dev, d.net} { + if srv == nil { + continue + } + if err := srv.Shutdown(shutdownCtx); err != nil && drainErr == nil { + drainErr = err + } + } + // Every door is drained, so no further frame can arrive: the hub's counters + // are final. Report them — this is the only non-test reader of the frame-loss + // accounting, and without it a run that committed none of the agent's + // conversation would end indistinguishably from one that committed all of it. + logFrameDiagnostics(shutdownCtx, d.log, d.hub) + if drainErr != nil { + return fmt.Errorf("draining compass.v1 servers on shutdown: %w", drainErr) + } + return nil +} + // publishReady stamps the initial ServerStatus{Ready} onto the bus. func publishReady(bus *events.Bus[busPayload]) { bus.Publish(&compassv1.SubscribeEventsResponse{ diff --git a/go/server/sinks.go b/go/server/sinks.go index 8341c4f5..a9ce98c6 100644 --- a/go/server/sinks.go +++ b/go/server/sinks.go @@ -6,26 +6,26 @@ // (runnerhub/hub.go): a lifecycle transition to the board, a conversation frame // to comms, a session-trace frame to the observation pane. // -// T4 wires the lifecycle sink for real — an agent-session state transition fans -// onto SubscribeEvents, the board/liveness surface the T4 acceptance path tails -// (provision a workspace, observe lifecycle on SubscribeEvents). The conversation -// and session-tail sinks are minimal in T4 by the frozen design -// (compass-0.6 §T4/§T5): the agent that emits conversation and trace frames is -// the first-party agent built in T5, and the conversation write-through -// additionally needs a session→channel mapping that lands with T5's store -// schema. Until then no agent produces those frames on this path; a frame that -// nonetheless arrives is observed (logged) rather than dropped, and T5 -// substitutes the real comms write-through and the SubscribeAgentSession stream -// behind the same interfaces without touching the hub. +// All three sinks are real. The lifecycle sink is the Bridge board — an +// agent-session state transition fans onto SubscribeEvents, the board/liveness +// surface. The conversation sink is the comms write-through (commsConversationSink +// below), committing a relayed conversation frame to durable Message rows + +// SubscribeComms. The session-tail sink is the per-session fan-out backing +// SubscribeAgentSession. package server import ( "context" + "errors" "log/slog" + "connectrpc.com/connect" + compassv1 "github.com/sealedsecurity/compass/go/gen/compass/v1" "github.com/sealedsecurity/compass/go/internal/board" + "github.com/sealedsecurity/compass/go/internal/comms" "github.com/sealedsecurity/compass/go/internal/runnerhub" + "github.com/sealedsecurity/compass/go/internal/store" ) // The lifecycle sink is the Bridge board (internal/board): a session lifecycle @@ -34,27 +34,77 @@ import ( // the live stream carried. The board is a strict superset of a bus-only sink, so // it is the one lifecycle sink — see newRunnerHub. -// observedConversationSink is the T4-minimal ConversationSink. The real -// write-through commits an agent conversation frame to comms Message rows + -// SubscribeComms; it needs the session→channel mapping T5's store adds and a -// first-party agent (T5) to emit the frames, so no agent produces conversation -// frames on this path in T4. A frame that nonetheless arrives is logged (never -// silently dropped) so a contract skew is observable; T5 replaces this with the -// comms write-through behind the same interface. -type observedConversationSink struct { - log *slog.Logger +// commsConversationSink is the real conversation write-through: a relayed +// conversation frame becomes a durable comms Message row and fans out on +// SubscribeComms, so an agent's turn is indistinguishable downstream from a +// human's post (SEA-1364 T3). +// +// It is a thin adapter, deliberately. All the work lives in the CommsService +// handler's *AsAccount family (internal/comms/agent_caller.go): a post delegates +// to the same PostMessage path a human takes, and an update goes through the +// authorizing store update that requires channel membership AND authorship. So +// there is no server-package authz to drift from the comms package's — this type +// only maps the sink interface onto those two calls. +// +// The account arrives already resolved: the hub owns the session->account +// binding and resolves it at the Deliver site, so this sink never looks a +// session up (runnerhub/hub.go, ConversationSink). +type commsConversationSink struct { + comms *comms.Comms } -func (s observedConversationSink) PostAgentMessage(_ context.Context, sessionID string, _ *compassv1.MessagePosted, updated *compassv1.MessageUpdated) error { - kind := "posted" - if updated != nil { - kind = "updated" +// PostAgentMessage commits one relayed conversation frame under account. +// +// ERROR VOCABULARY. The hub classifies what this returns by its CONNECT CODE +// alone (runnerhub/hub.go, ConversationSink) — NotFound/InvalidArgument drop the +// frame as a refusal, FailedPrecondition drops it as a contract defect, anything +// else ends the Runner's relay stream. This method satisfies that contract by +// construction: both delegates return errors already mapped through the comms +// package's edgeError, and the one error minted here +// (errNoConversationVariant) picks its code deliberately. Never return a bare +// error from here — connect.CodeOf reports CodeUnknown for one, which tears the +// stream down. +// +// A frame with neither variant set cannot happen: Deliver dispatches on the +// oneof and only ever passes exactly one. The unset case is therefore a guard, +// returning an error rather than panicking or reporting a silent success +// (rule://go-no-panic-in-lib). +func (s commsConversationSink) PostAgentMessage( + ctx context.Context, + account store.AccountID, + _ string, + posted *compassv1.MessagePosted, + updated *compassv1.MessageUpdated, +) error { + switch { + case posted != nil: + _, err := s.comms.CommitAgentPost(ctx, account, posted) + return err + case updated != nil: + _, err := s.comms.CommitAgentUpdate(ctx, account, updated) + return err + default: + return errNoConversationVariant } - s.log.Debug("conversation frame relayed before the T5 comms write-through is wired; observed, not committed", - slog.String("session_id", sessionID), slog.String("kind", kind)) - return nil } +// errNoConversationVariant is the cause for a conversation write-through called +// with neither variant set. Unreachable through Deliver's dispatch, so it can +// only fire on a SERVER-SIDE wiring defect — the frame is well-formed and the +// dispatch is what broke. +// +// CodeInternal, so the hub ends the relay stream. It was CodeInvalidArgument, +// which classified a Server bug as a routine droppable frame: the Server would +// have silently discarded relayed turns because of its own defect, hidden among +// the refusals a healthy relay is expected to produce. An unreachable branch +// firing is exactly the case worth a loud teardown. It is deliberately not +// CodeFailedPrecondition either — that bucket is for a skew the relay keeps +// serving through, and this one must not be served through. +var errNoConversationVariant = connect.NewError( + connect.CodeInternal, + errors.New("server: conversation frame has neither a posted nor an updated variant"), +) + // newRunnerHub's session-tail sink is the real per-session fan-out (sessionTail, // sessiontail.go): RelaySessionFrame repackages each internal frame to its // public form and fans it to that session's live SubscribeAgentSession @@ -65,21 +115,56 @@ func (s observedConversationSink) PostAgentMessage(_ context.Context, sessionID // newRunnerHub constructs the Server-side RunnerHub over its write-through sinks // and the agent-comms caller: the Bridge board as the lifecycle sink (a session // transition is recorded into the board projection and fanned onto -// SubscribeEvents), the minimal conversation sink (see the file doc), the real +// SubscribeEvents), the comms write-through as the conversation sink (a relayed +// conversation frame becomes a durable Message row + SubscribeComms), the real // per-session tail sink for SubscribeAgentSession passed in by serve.go so the // same instance is shared with the service, and comms — the CommsService handler, // which executes an agent-initiated comms call under the account a session -// resolves to (RelayCommsCall). log carries the sinks' observed-frame diagnostics -// and the hub's gap/unknown-frame warnings; nil falls back to slog.Default(). -func newRunnerHub(brd *board.Projection, tail runnerhub.SessionTailSink, comms runnerhub.CommsCaller, log *slog.Logger) *runnerhub.Hub { +// resolves to (RelayCommsCall). The one CommsService instance serves both comms +// legs: the conversation write-through and RelayCommsCall. log carries the hub's +// gap/unknown/refused-frame diagnostics; nil falls back to slog.Default(). +func newRunnerHub(brd *board.Projection, tail runnerhub.SessionTailSink, commsSvc *comms.Comms, log *slog.Logger) *runnerhub.Hub { if log == nil { log = slog.Default() } return runnerhub.NewHub( - observedConversationSink{log: log}, + commsConversationSink{comms: commsSvc}, brd, tail, - comms, + commsSvc, log, ) } + +// logFrameDiagnostics emits the hub's frame-loss snapshot as one line. Serve +// calls it on shutdown, so every run states plainly how many relayed frames +// never reached their surface. +// +// It exists because the counters had no non-test reader at all. Total frame loss +// was observable only by reading per-frame warn lines and tallying them by hand +// — which means, in practice, that a relay committing NOTHING looked exactly +// like a relay committing everything to anyone not already suspicious. That is +// the failure mode this whole classification split exists to expose, so leaving +// the numbers unreadable would have undone it. +// +// A clean run logs at Info and reads as four zeros. Any contract defect flips it +// to Error and says the relay is misconfigured, because on the current base a +// non-zero defect count means every agent turn was lost (runnerhub/hub.go, +// ContractDefects). This is the minimum that makes the counters observable — the +// server has no metrics surface to register a gauge on, and inventing one is not +// this change's job. A future metrics or diagnostics RPC reads the same +// FrameDiagnostics snapshot. +func logFrameDiagnostics(ctx context.Context, log *slog.Logger, hub *runnerhub.Hub) { + d := hub.FrameDiagnostics() + attrs := []any{ + slog.Uint64("contract_defects", d.ContractDefects), + slog.Uint64("refused_frames", d.RefusedFrames), + slog.Uint64("unknown_frames", d.UnknownFrames), + slog.Bool("seen_sequence_gap", d.SeenGap), + } + if d.ContractDefects > 0 { + log.ErrorContext(ctx, "relayed agent frames were dropped for relay misconfiguration: agent conversation was NOT committed to comms", attrs...) + return + } + log.InfoContext(ctx, "relayed agent frame accounting", attrs...) +} diff --git a/go/server/sinks_test.go b/go/server/sinks_test.go new file mode 100644 index 00000000..9e52a176 --- /dev/null +++ b/go/server/sinks_test.go @@ -0,0 +1,56 @@ +//go:build unix + +package server + +// The conversation sink's error vocabulary. commsConversationSink is the one +// production ConversationSink, and the RunnerHub decides a relayed frame's fate +// purely from the CONNECT CODE this type returns (runnerhub/hub.go, +// ConversationSink's error-vocabulary contract): NotFound/InvalidArgument drop +// the frame and count it, FailedPrecondition drops it as a contract defect, +// anything else tears the Runner's PublishEvents stream down. +// +// So a code chosen here is not cosmetic — it is the difference between a frame +// quietly discarded and an operator finding out. This pins the one code this +// file mints itself. The other half of the linkage (that CodeInternal is what +// the hub tears down on, and that a bare unmapped error does too) is pinned on +// the hub side, where the predicates live: +// runnerhub.TestDeliverConversationStoreFailureTearsDownTheStream and +// TestDeliverBareSinkErrorTearsDownTheStreamAndCountsNothing. + +import ( + "context" + "testing" + + "connectrpc.com/connect" +) + +// The no-variant guard tears the stream DOWN rather than being dropped as a +// refusal. Its own docstring says it is unreachable through Deliver's dispatch +// (Deliver switches on the oneof and passes exactly one variant), so if it ever +// fires, a Server-side dispatch defect has been introduced — the frame is fine +// and the Server is broken. +// +// CodeInvalidArgument, which this returned before, put it in the routine +// droppable bucket: the Server would have silently discarded frames on its own +// wiring bug, at the one moment loudness is worth the teardown. It must not be +// CodeFailedPrecondition either — that is the non-fatal contract-defect bucket, +// for a skew the relay keeps serving through, and an unreachable Server branch +// firing is not something to keep serving through. +// +// Mutation: revert to CodeInvalidArgument (or FailedPrecondition) → the frame +// becomes a silent drop on a Server bug, and this reddens. +func TestNoConversationVariantIsAnInternalDefectNotADroppableRefusal(t *testing.T) { + // The sink is called with neither variant: the shape only a dispatch bug + // produces. A nil comms handler is safe precisely because this branch must + // return before touching it — reaching comms here would panic, which is + // itself the assertion that the guard fires first rather than dereferencing. + sink := commsConversationSink{comms: nil} + + err := sink.PostAgentMessage(context.Background(), "acct-agent", "sess-1", nil, nil) + if err == nil { + t.Fatal("PostAgentMessage(no variant) = nil, want an error — a silent success would ack a frame that committed nothing") + } + if got := connect.CodeOf(err); got != connect.CodeInternal { + t.Fatalf("PostAgentMessage(no variant) code = %v, want %v — an unreachable branch firing is a Server-side wiring defect, and it must tear the stream down rather than hide among expected per-frame refusals", got, connect.CodeInternal) + } +} From 53dc634bc3ecac95a7742414f1492108d42620e8 Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 29 Jul 2026 14:08:02 -0400 Subject: [PATCH 2/2] fix(compass): reconcile ask_id on relayed conversation UPDATE write-through The UPDATE write-through mapped frames through blocksFromWire -> askFromWire, which unconditionally stripped ask_id (correct for POST, where the server mints via mintAskIDs, but wrong for UPDATE). UpdateMessageBlocksAsAuthor then rejected the id-less ask, so any ask-bearing update was refused and never persisted. Add an UPDATE-path mapper (updateBlocksFromWire / askFromWireForUpdate) that preserves the wire ask_id, and reconcile it at the comms edge against the stored row before the write: ask_id is immutable once minted, so a scoped, message-id read (store.MessageAskIDs) is race-free and does not touch the store's single-statement authz gate. An id-less update ask is filled from the stored id; a non-empty wire ask_id that disagrees with the stored one is rejected as a distinct, clearly-messaged forged-frame error (still InvalidArgument), never conflated with the generic id-less case. Co-Authored-By: seal --- go/internal/comms/agent_caller.go | 118 +++++++++- .../comms/agent_conversation_pgtest_test.go | 203 +++++++++++++++++- go/internal/comms/mapping.go | 73 ++++--- go/internal/store/messages.go | 40 ++++ .../store/messages_authored_update_test.go | 57 ++++- 5 files changed, 451 insertions(+), 40 deletions(-) diff --git a/go/internal/comms/agent_caller.go b/go/internal/comms/agent_caller.go index 0d96e69c..fb7e7d7b 100644 --- a/go/internal/comms/agent_caller.go +++ b/go/internal/comms/agent_caller.go @@ -33,6 +33,7 @@ package comms import ( "context" + "fmt" "connectrpc.com/connect" @@ -79,6 +80,40 @@ func (notAgentAccountError) Error() string { return "comms: agent-initiated call resolved to a non-agent account, which has no home channel; the session->account binding is wrong" } +// askIDMismatchError is the cause when a relayed UPDATE frame's ask block carries +// a non-empty ask_id that DISAGREES with the id stored for that block — a +// forged/malformed frame, distinct from an ask block that simply arrived id-less +// (which reconcileUpdateAskIDs fills from the stored row). It maps to +// CodeInvalidArgument, but its own message so the mismatch reads as the specific +// defect it is, never conflated with the generic "no ask_id" refusal. +type askIDMismatchError struct { + block int + wire string + stored string +} + +func (e askIDMismatchError) Error() string { + return fmt.Sprintf( + "comms: block %d ask_id %q from the update frame does not match the stored ask_id %q; ask_id is immutable and an update must carry the stored id, not a different one", + e.block, e.wire, e.stored) +} + +// surplusAskError is the cause when a relayed UPDATE frame carries MORE ask +// blocks than the stored message has — a surplus ask with no stored counterpart +// to reconcile against. An update cannot introduce a new ask (a fresh ask_id is +// minted only on POST), so a surplus ask is always illegitimate: rejecting it +// keeps a caller-chosen ask_id off the UPDATE path, the same no-forged-ask_id +// invariant askFromWire enforces for POST. Maps to CodeInvalidArgument. +type surplusAskError struct { + block int +} + +func (e surplusAskError) Error() string { + return fmt.Sprintf( + "comms: block %d is an ask the stored message does not have; an update cannot introduce a new ask (ask_id is minted only when the ask is first posted)", + e.block) +} + // PostAsAccount executes one agent-initiated PostMessage as account. It sets the // account on the context (WithActor) and delegates to the same PostMessage // handler path a human caller takes, so authz (D9), idempotency @@ -202,9 +237,19 @@ func (c *Comms) CommitAgentPost( // // Errors map through edgeError to the same Connect codes a human caller gets: // CodeNotFound for a refused row, CodeInvalidArgument for a malformed frame (an -// empty message.id, an empty block set, an ask block missing its immutable -// ask_id). Those are the refusals the hub treats as non-fatal drops rather than -// stream teardowns. +// empty message.id, an empty block set, or an ask block whose ask_id cannot be +// reconciled against the stored row — see reconcileUpdateAskIDs). Those are the +// refusals the hub treats as non-fatal drops rather than stream teardowns. +// +// Ask_id handling is the one place the UPDATE frame differs from a POST. A POST +// strips any wire ask_id and lets the store mint one; an update must instead +// carry back the id the append already minted, so this path maps blocks with +// updateBlocksFromWire (which PRESERVES the wire ask_id) and then reconciles that +// id against the stored row before the write. The stored ask_id is authoritative +// (it is immutable once minted), so reconciliation fills an id-less update ask +// from the stored ask and rejects a non-empty wire ask_id that disagrees with +// the stored one as a forged/malformed frame — never blindly trusting the wire +// value, which would reopen the forgery/collision risk askFromWire guards. func (c *Comms) CommitAgentUpdate( ctx context.Context, account store.AccountID, @@ -214,10 +259,13 @@ func (c *Comms) CommitAgentUpdate( return nil, errNoActor } msg := updated.GetMessage() - blocks, err := blocksFromWire(msg.GetBlocks()) + blocks, err := updateBlocksFromWire(msg.GetBlocks()) if err != nil { return nil, err } + if err := c.reconcileUpdateAskIDs(ctx, store.MessageID(msg.GetId()), blocks); err != nil { + return nil, err + } stored, err := c.store.UpdateMessageBlocksAsAuthor(ctx, account, store.MessageID(msg.GetId()), blocks) if err != nil { return nil, edgeError(err) @@ -229,6 +277,68 @@ func (c *Comms) CommitAgentUpdate( return &compassv1.MessageUpdated{Message: messageToWire(stored)}, nil } +// reconcileUpdateAskIDs makes the ask blocks of a relayed UPDATE frame carry the +// stored, server-owned ask_id — the safe alternative to trusting the wire value. +// It reads the stored row's ask_ids (an immutable field, so a separate read from +// the authz UPDATE that follows is race-free — see store.MessageAskIDs) and, for +// the k-th ask block of the frame, reconciles it against the k-th stored ask: +// +// - an id-LESS update ask is filled from the stored ask_id (the common case — +// the id the append minted, which the store then requires); +// - a non-empty wire ask_id that MATCHES the stored one is left as-is; +// - a non-empty wire ask_id that DISAGREES with the stored one is a +// forged/malformed frame, rejected CodeInvalidArgument and distinctly +// messaged (never conflated with the generic id-less case, and never +// silently overwriting the stored id). +// +// Matching is POSITIONAL among ask blocks: a streaming turn re-sends its FULL +// current block set in stable order (comms.proto:388-390), so the k-th ask of +// the frame is the k-th ask of the row. If the frame carries MORE ask blocks +// than the stored row, the surplus ask has no stored counterpart and is +// REJECTED (CodeInvalidArgument): an update cannot introduce a brand-new ask — +// a fresh ask is minted only on POST — so a surplus ask is always illegitimate, +// whether id-less (unanswerable) or carrying a caller-chosen id (the forgery the +// POST path strips to keep RespondToAsk's containment SELECT unambiguous). The +// store guards only the empty-id case, so rejecting here is what closes the +// non-empty forged-id surplus. +func (c *Comms) reconcileUpdateAskIDs(ctx context.Context, id store.MessageID, blocks []store.MessageBlock) error { + storedAskIDs, err := c.store.MessageAskIDs(ctx, id) + if err != nil { + return edgeError(err) + } + askIdx := 0 + for i := range blocks { + ask := blocks[i].Ask + if ask == nil { + continue + } + if askIdx >= len(storedAskIDs) { + // A surplus ask block (beyond the stored ask count) has no stored + // counterpart to reconcile against. An UPDATE cannot legitimately + // introduce a new ask — a fresh ask is minted only on the POST path + // (mintAskIDs) — so any surplus ask is illegitimate regardless of its + // ask_id: an id-less one could not be answered (no minted id), and a + // NON-empty one is a caller-chosen id, exactly the forgery the POST + // path strips to prevent (askFromWire: a shared ask_id makes + // RespondToAsk's containment SELECT match multiple rows). Reject it + // here rather than passing a wire id through to the store, which only + // guards the empty case. + return connect.NewError(connect.CodeInvalidArgument, + surplusAskError{block: i}) + } + stored := storedAskIDs[askIdx] + switch { + case ask.AskID == "": + ask.AskID = stored + case ask.AskID != stored: + return connect.NewError(connect.CodeInvalidArgument, + askIDMismatchError{block: i, wire: ask.AskID, stored: stored}) + } + askIdx++ + } + return nil +} + // defaultChannel returns req with an empty channel_id filled from the account's // home channel, so the common "post in my own channel" case needs no id plumbed // into the container. A non-empty channel_id is left untouched. The returned diff --git a/go/internal/comms/agent_conversation_pgtest_test.go b/go/internal/comms/agent_conversation_pgtest_test.go index ea260c1d..13c22bf2 100644 --- a/go/internal/comms/agent_conversation_pgtest_test.go +++ b/go/internal/comms/agent_conversation_pgtest_test.go @@ -25,6 +25,7 @@ package comms import ( "context" + "strings" "testing" "connectrpc.com/connect" @@ -46,12 +47,21 @@ func updatedFrame(id string, blocks []*compassv1.MessageBlock) *compassv1.Messag return &compassv1.MessageUpdated{Message: &compassv1.Message{Id: id, Blocks: blocks}} } -// askBlockWire builds a wire ask block carrying one question, for the malformed -// -ask rejection case (the id an update must carry is the store's, minted at -// append; a wire ask always enters id-less, so an UPDATE carrying one is -// necessarily missing its immutable id). +// askBlockWire builds a wire ask block carrying one question and NO ask_id — the +// id-less shape a genuinely malformed update frame carries (an update must carry +// back the id the append minted; an ask block with no id at all is a malformed +// update, distinct from one whose id merely disagrees with the stored row). func askBlockWire() *compassv1.MessageBlock { + return askBlockWireID("") +} + +// askBlockWireID builds a wire ask block carrying one question and the given +// ask_id — used to relay an UPDATE that carries the id the append minted (the +// legitimate ask-bearing update) or a forged one that disagrees with the stored +// row. +func askBlockWireID(askID string) *compassv1.MessageBlock { return &compassv1.MessageBlock{Block: &compassv1.MessageBlock_Ask{Ask: &compassv1.Ask{ + AskId: askID, Questions: []*compassv1.AskQuestion{{ QuestionId: "q1", Question: "Which environment?", @@ -291,7 +301,13 @@ func TestCommitAgentUpdateRevokedMemberIsNotFound(t *testing.T) { // The two malformed-input refusals, both CodeInvalidArgument: an update whose // message.id is empty (a frame the agent never stamped) and an update carrying -// an ask block with no ask_id (which would orphan a pending RespondToAsk). +// an ask block with NO ask_id that has no stored counterpart to reconcile it +// from — a genuinely id-less frame, distinct from an ask-bearing update carrying +// its stored id (which now PERSISTS, see TestCommitAgentUpdatePersistsAskBlock) +// and from one whose id merely disagrees with the stored row (see +// TestCommitAgentUpdateRejectsAskIDMismatch). An id-less ask that cannot be +// reconciled is still an error, because the store's immutable-ask_id contract +// requires an update to carry the id the append minted. func TestCommitAgentUpdateRejectsMalformedFrames(t *testing.T) { svc, st := newHandler(t) ctx := context.Background() @@ -308,9 +324,12 @@ func TestCommitAgentUpdateRejectsMalformedFrames(t *testing.T) { connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(empty message id)") }) t.Run("ask block with no ask id", func(t *testing.T) { - // A wire ask always enters id-less (askFromWire strips any caller value), - // so an UPDATE carrying one is necessarily missing the immutable id the - // store minted at append. + // The seeded row is text-only, so an ask block on the update has NO + // stored counterpart for reconcileUpdateAskIDs to fill its id from: it + // stays id-less and the store rejects it. This pins the truly id-less + // case (an update that cannot carry the immutable id the append would + // have minted), NOT that ask-bearing updates are unsupported — one + // carrying its stored id persists (TestCommitAgentUpdatePersistsAskBlock). _, err := svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(posted.GetMessage().GetId(), []*compassv1.MessageBlock{askBlockWire()})) connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(ask with no ask_id)") @@ -499,3 +518,171 @@ func TestAgentCallsWithANonAgentAccountAreRefusedNotPanics(t *testing.T) { _, err = svc.ListAsAccount(ctx, user.ID, &compassv1.ListMessagesRequest{}) connectCodeIs(t, err, connect.CodeFailedPrecondition, "ListAsAccount(user account, empty channel)") } + +// THE positive contract this fix restores: an addressed conversation UPDATE +// carrying an ask block WITH the stored ask_id PERSISTS, rather than being +// refused. Before the fix, updateBlocksFromWire did not exist and the update +// went through blocksFromWire -> askFromWire, which unconditionally stripped the +// ask_id, so UpdateMessageBlocksAsAuthor rejected the id-less ask as +// CodeInvalidArgument and the update never committed. Now the update path +// preserves the wire ask_id and reconciles it against the stored row, so the +// ask survives the write and round-trips its id. +// +// Mutation that reddens it: routing the update back through blocksFromWire (the +// POST mapper) reintroduces the strip and the ask is refused again. +func TestCommitAgentUpdatePersistsAskBlock(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + + // Seed a real ask through the POST path so the store mints the ask_id the + // update must carry back. The minted id is read off the committed row. + posted, err := svc.CommitAgentPost(ctx, agent.ID, + postedFrame([]*compassv1.MessageBlock{askBlockWire()})) + if err != nil { + t.Fatalf("CommitAgentPost(ask): %v", err) + } + id := posted.GetMessage().GetId() + storedAskID := posted.GetMessage().GetBlocks()[0].GetAsk().GetAskId() + if storedAskID == "" { + t.Fatalf("post did not mint an ask_id; got %+v", posted.GetMessage().GetBlocks()) + } + + // The streaming turn re-sends the full block set: a new text block plus the + // SAME ask, carrying the id the append minted. This is the frame that used + // to be refused. + updated, err := svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, + []*compassv1.MessageBlock{ + {Block: &compassv1.MessageBlock_Text{Text: "settled turn"}}, + askBlockWireID(storedAskID), + })) + if err != nil { + t.Fatalf("CommitAgentUpdate(ask with stored ask_id) was refused, want persisted: %v", err) + } + if got := updated.GetMessage().GetBlocks(); len(got) != 2 { + t.Fatalf("updated message has %d blocks, want 2 (text + ask)", len(got)) + } else if got[1].GetAsk().GetAskId() != storedAskID { + t.Fatalf("updated ask_id = %q, want the preserved %q", got[1].GetAsk().GetAskId(), storedAskID) + } + + // Durable: the row on the home channel carries the updated set and the same + // ask_id, so a pending RespondToAsk still correlates. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("home channel holds %d messages, want 1 (in-place edit)", len(msgs)) + } + blocks := msgs[0].Blocks + if len(blocks) != 2 || blocks[0].Text == nil || *blocks[0].Text != "settled turn" { + t.Fatalf("stored blocks = %+v, want [text 'settled turn', ask]", blocks) + } + if blocks[1].Ask == nil || blocks[1].Ask.AskID != storedAskID { + t.Fatalf("stored ask_id = %v, want the preserved %q", blocks[1].Ask, storedAskID) + } +} + +// The distinct misclassification the review flagged: an ask-bearing update that +// carries a NON-EMPTY ask_id disagreeing with the stored one is a forged/ +// malformed frame, refused CodeInvalidArgument — and refused by +// reconcileUpdateAskIDs's own mismatch error, never conflated with the generic +// id-less case and never silently overwriting the stored id. It stays +// CodeInvalidArgument (still a caller error), but a distinct, clearly-messaged +// one. +// +// Mutation that reddens it: blindly trusting the wire ask_id (dropping the +// mismatch branch) lets the forged id through and the refusal disappears. +func TestCommitAgentUpdateRejectsAskIDMismatch(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + posted, err := svc.CommitAgentPost(ctx, agent.ID, + postedFrame([]*compassv1.MessageBlock{askBlockWire()})) + if err != nil { + t.Fatalf("CommitAgentPost(ask): %v", err) + } + id := posted.GetMessage().GetId() + storedAskID := posted.GetMessage().GetBlocks()[0].GetAsk().GetAskId() + _, err = svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, + []*compassv1.MessageBlock{askBlockWireID("forged-" + storedAskID)})) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(ask_id mismatch)") + // The classification must be the DISTINCT mismatch, not the generic id-less + // refusal ("ask has no ask_id"): asserting the code alone would pass against + // the old strip-everything behavior too, which is the misclassification the + // review flagged. Pinning the message keeps the two errors distinct. + if err == nil || !strings.Contains(err.Error(), "does not match the stored ask_id") { + t.Fatalf("mismatch error = %v, want the distinct 'does not match the stored ask_id' classification", err) + } + + // The forged update left no trace: the row still carries the original ask + // with its minted id, so RespondToAsk against that id still resolves. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 || len(msgs[0].Blocks) != 1 || msgs[0].Blocks[0].Ask == nil { + t.Fatalf("stored blocks = %+v, want the single untouched ask", msgs[0].Blocks) + } + if msgs[0].Blocks[0].Ask.AskID != storedAskID { + t.Fatalf("stored ask_id = %q, want the untouched %q", msgs[0].Blocks[0].Ask.AskID, storedAskID) + } +} + +// The forgery vector the review flagged: a relayed UPDATE that appends a SURPLUS +// ask block (beyond the stored ask count) carrying a caller-chosen non-empty +// ask_id must be refused, not persisted. A surplus ask has no stored counterpart +// to reconcile against, and an update cannot introduce a new ask (ask_id is +// minted only on POST), so accepting a caller-supplied id would reopen exactly +// the collision askFromWire strips to prevent on POST: a forged id shared with +// another message makes RespondToAsk's containment SELECT match both rows. The +// store guards only the empty-id case, so the edge must reject the non-empty +// surplus. +// +// Mutation that reddens it: skipping surplus asks in reconcileUpdateAskIDs (the +// pre-fix `if askIdx < len(storedAskIDs)` gate) lets the forged id through and +// the row is silently updated with it. +func TestCommitAgentUpdateRejectsSurplusForgedAsk(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + + owner := mustUser(t, st, "owner") + agent := mustAgent(t, st, owner.ID, "agent") + posted, err := svc.CommitAgentPost(ctx, agent.ID, + postedFrame([]*compassv1.MessageBlock{askBlockWire()})) + if err != nil { + t.Fatalf("CommitAgentPost(ask): %v", err) + } + id := posted.GetMessage().GetId() + storedAskID := posted.GetMessage().GetBlocks()[0].GetAsk().GetAskId() + + // Re-send the stored ask (correctly carrying its minted id) AND a surplus + // ask carrying a forged, caller-chosen id. The first reconciles cleanly; the + // surplus has no stored counterpart and must be rejected. + _, err = svc.CommitAgentUpdate(ctx, agent.ID, updatedFrame(id, + []*compassv1.MessageBlock{ + askBlockWireID(storedAskID), + askBlockWireID("forged-surplus-ask"), + })) + connectCodeIs(t, err, connect.CodeInvalidArgument, "CommitAgentUpdate(surplus forged ask)") + if err == nil || !strings.Contains(err.Error(), "an update cannot introduce a new ask") { + t.Fatalf("surplus error = %v, want the distinct 'an update cannot introduce a new ask' classification", err) + } + + // The forged surplus left no trace: the row still carries only the original + // ask with its minted id, and the forged id was never persisted. + msgs, err := st.ListMessages(ctx, agent.ID, store.ContainerRef{ChannelID: agent.Agent.HomeChannelID}, store.Page{}) + if err != nil { + t.Fatalf("ListMessages: %v", err) + } + if len(msgs) != 1 || len(msgs[0].Blocks) != 1 || msgs[0].Blocks[0].Ask == nil { + t.Fatalf("stored blocks = %+v, want the single untouched ask", msgs[0].Blocks) + } + if msgs[0].Blocks[0].Ask.AskID != storedAskID { + t.Fatalf("stored ask_id = %q, want the untouched %q", msgs[0].Blocks[0].Ask.AskID, storedAskID) + } +} diff --git a/go/internal/comms/mapping.go b/go/internal/comms/mapping.go index 4a8d7ad1..e05f428f 100644 --- a/go/internal/comms/mapping.go +++ b/go/internal/comms/mapping.go @@ -249,7 +249,54 @@ func blocksFromWire(blocks []*compassv1.MessageBlock) ([]store.MessageBlock, err return out, nil } +// updateBlocksFromWire maps wire blocks for the UPDATE write-through +// (CommitAgentUpdate), differing from blocksFromWire in ONE respect: an ask +// block PRESERVES its wire ask_id rather than stripping it. Stripping is correct +// for a POST (the server mints a fresh id via mintAskIDs), but an update carries +// the id the append already minted — dropping it would make every ask-bearing +// update fail the store's immutable-ask_id check. The preserved value is NOT +// trusted as-is: CommitAgentUpdate reconciles it against the stored row before +// the write (an empty id is filled from the stored ask, a non-empty id that +// disagrees with the stored one is rejected as a forged frame). +func updateBlocksFromWire(blocks []*compassv1.MessageBlock) ([]store.MessageBlock, error) { + out := make([]store.MessageBlock, 0, len(blocks)) + for _, b := range blocks { + switch body := b.GetBlock().(type) { + case *compassv1.MessageBlock_Text: + text := body.Text + out = append(out, store.MessageBlock{Text: &text}) + case *compassv1.MessageBlock_Ask: + out = append(out, store.MessageBlock{Ask: askFromWireForUpdate(body.Ask)}) + default: + return nil, connect.NewError(connect.CodeInvalidArgument, + errEmptyBlock) + } + } + return out, nil +} + func askFromWire(a *compassv1.Ask) *store.Ask { + // ask_id is server-owned on the POST path: drop any caller-supplied value so + // the store's mintAskIDs always assigns a fresh, globally-unique id on append + // (comms.proto: "server-assigned and globally unique"). Honoring a wire value + // would let two posts share an ask_id, making RespondToAsk's containment + // SELECT match multiple rows and answer a nondeterministic one. + return &store.Ask{AskID: "", Questions: askQuestionsFromWire(a)} +} + +// askFromWireForUpdate is askFromWire's UPDATE sibling: it PRESERVES the wire +// ask_id (an update must carry back the id minted at append) instead of +// stripping it. Safe only because CommitAgentUpdate reconciles the preserved +// value against the immutable stored ask_id before writing — this mapper alone +// does not trust it. +func askFromWireForUpdate(a *compassv1.Ask) *store.Ask { + return &store.Ask{AskID: a.GetAskId(), Questions: askQuestionsFromWire(a)} +} + +// askQuestionsFromWire maps an ask's questions onto store types, shared by the +// POST (askFromWire) and UPDATE (askFromWireForUpdate) mappers — the two differ +// only in ask_id handling, never in how the question set crosses the edge. +func askQuestionsFromWire(a *compassv1.Ask) []store.AskQuestion { questions := make([]store.AskQuestion, len(a.GetQuestions())) for i, q := range a.GetQuestions() { opts := make([]store.AskOption, len(q.GetOptions())) @@ -265,31 +312,7 @@ func askFromWire(a *compassv1.Ask) *store.Ask { Recommended: q.Recommended, } } - // ask_id and every answer-state field are server-owned: drop any - // caller-supplied value. - // - // ask_id, because the store's mintAskIDs must assign a fresh globally-unique - // id on append (comms.proto: "server-assigned and globally unique"). - // Honoring a wire value would let two posts share an ask_id, making - // RespondToAsk's containment SELECT match multiple rows and answer a - // nondeterministic one. - // - // The answer state — Answered, and the per-question ChosenOptionIDs / - // CustomText / TimedOut — because an ask arriving over the wire has by - // definition not been answered yet: it is being posted, and only - // RespondToAsk may record an answer. Honoring them let a caller post an ask - // that arrives pre-populated, which a client reading the per-question - // fields renders as already settled while the server still holds it - // pending and answerable. - // - // Omission is the mechanism, but it is NOT self-enforcing: a keyed composite - // literal compiles and vets clean with any subset of fields set, so a field - // added to store.Ask or store.AskQuestion is silently honored here from - // whatever the caller sent. Every new field must therefore be consciously - // classified — content (map it) or answer state (omit it) — and - // TestAskFromWireDropsEveryServerOwnedField walks the descriptor to fail - // when one is added without that decision. - return &store.Ask{AskID: "", Questions: questions} + return questions } // ---- write-through event publishers ---- diff --git a/go/internal/store/messages.go b/go/internal/store/messages.go index 2b369d88..c34fd721 100644 --- a/go/internal/store/messages.go +++ b/go/internal/store/messages.go @@ -268,6 +268,46 @@ func (s *Store) UpdateMessageBlocksAsAuthor(ctx context.Context, actor AccountID return msgs[0], nil } +// MessageAskIDs returns the ask_id of every ask block on the message, in block +// order, for the relayed-update write-through's ask_id reconciliation +// (comms.CommitAgentUpdate). It exists so the UPDATE path can source the +// server-owned ask_id from the stored row instead of stripping it (the POST +// path's mintAskIDs behavior, wrong for an update) or trusting a wire value. +// +// Safe as a SEPARATE statement from the authz UPDATE precisely because ask_id is +// immutable: mintAskIDs (blocks.go) assigns it once at append and nothing ever +// reassigns it, so a value read here cannot be invalidated by a later write — +// unlike the mutable post-state the UPDATE returns via RETURNING, this read +// observes a field that is stable for the row's life. It is scoped by message id +// ALONE — the same scope the UPDATE addresses — and performs NO membership or +// authorship check and returns NO distinct not-found (an unknown id or a message +// with no ask yields an empty slice), so it cannot be turned into an +// authz/session enumeration oracle: the sole authz gate remains the +// single-statement UpdateMessageBlocksAsAuthor that follows, and its result is +// never derived from what this read returned. +func (s *Store) MessageAskIDs(ctx context.Context, id MessageID) ([]string, error) { + var blocksJSON []byte + if err := s.pool.QueryRow(ctx, + `SELECT blocks FROM messages WHERE id = $1`, string(id), + ).Scan(&blocksJSON); err != nil { + if noRows(err) { + return nil, nil + } + return nil, fmt.Errorf("store: read message ask ids: %w", err) + } + blocks, err := unmarshalBlocks(blocksJSON) + if err != nil { + return nil, err + } + var ids []string + for _, b := range blocks { + if b.Ask != nil { + ids = append(ids, b.Ask.AskID) + } + } + return ids, nil +} + // ListMessages pages a channel's messages newest-first, clamped to the store's // page bounds (comms.proto:446-461). Ordering keys on the monotonic seq (a // stable total order even under equal timestamps); BeforeMessageID pages diff --git a/go/internal/store/messages_authored_update_test.go b/go/internal/store/messages_authored_update_test.go index 5d868189..a3840378 100644 --- a/go/internal/store/messages_authored_update_test.go +++ b/go/internal/store/messages_authored_update_test.go @@ -180,7 +180,12 @@ func TestUpdateMessageBlocksAsAuthorUnknownMessageIsNotFound(t *testing.T) { // untouched. Empty message id is new here (updateMessageBlocksExec lets it fall // through to a zero-rows ErrNotFound); the empty block set and the empty AskID // mirror the checks the shared core already makes, restated because this path -// deliberately does not call it. +// deliberately does not call it. The empty-AskID case pins the id-LESS frame +// specifically — a genuinely malformed update that carries no id to persist — +// NOT that ask-bearing updates are unsupported: the store trusts and persists a +// caller-supplied NON-empty ask_id (TestUpdateMessageBlocksAsAuthorEditsInPlace +// AndReturnsRow already stores one), and the comms edge fills an id-less update +// ask from the stored row before it ever reaches here (reconcileUpdateAskIDs). func TestUpdateMessageBlocksAsAuthorRejectsMalformedInput(t *testing.T) { ctx := context.Background() s := newTestStore(t) @@ -201,8 +206,12 @@ func TestUpdateMessageBlocksAsAuthorRejectsMalformedInput(t *testing.T) { sentinelIs(t, err, ErrInvalidArgument, "empty block set") }) t.Run("ask block with no ask id", func(t *testing.T) { - // ask_id is minted once at append and immutable; an update carrying an - // id-less ask would orphan any pending RespondToAsk against the original. + // A truly id-LESS ask is still an error at the store: ask_id is minted + // once at append and immutable, so an update carrying no id has nothing + // to persist and would orphan any pending RespondToAsk against the + // original. This is the store's last-line guard; the comms edge reconciles + // an id-less UPDATE ask from the stored row before the write, so a + // legitimate ask-bearing update never arrives here id-less. _, err := s.UpdateMessageBlocksAsAuthor(ctx, author.ID, msg.ID, []MessageBlock{textBlock("x"), askBlockID("")}) sentinelIs(t, err, ErrInvalidArgument, "ask block with an empty ask_id") }) @@ -215,6 +224,48 @@ func TestUpdateMessageBlocksAsAuthorRejectsMalformedInput(t *testing.T) { assertBlocksEqual(t, got[0].Blocks, []MessageBlock{textBlock("untouched")}) } +// MessageAskIDs backs the comms edge's UPDATE ask_id reconciliation: it returns +// every ask block's stored ask_id in block order (an immutable field, so this +// separate read is race-free against the authz UPDATE that follows). A row with +// mixed text and ask blocks yields only the asks' ids, in order; an unknown id +// yields an empty slice rather than an error, so the read can never be turned +// into a not-found enumeration oracle. +// +// Mutation that reddens it: returning ids for text blocks too (breaks the +// positional match the edge relies on), or erroring on an unknown id (turns the +// benign read into a distinct not-found signal). +func TestMessageAskIDsReturnsStoredIDsInOrder(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + author := mustUser(t, s, "author") + ch := mustChannel(t, s, author.ID) + + msg, _, err := s.AppendMessage(ctx, Message{ + Container: ContainerRef{ChannelID: ch.ID}, AuthorAccountID: author.ID, + Blocks: []MessageBlock{askBlockID("ask-first"), textBlock("between"), askBlockID("ask-second")}, + }, "") + if err != nil { + t.Fatalf("AppendMessage: %v", err) + } + + ids, err := s.MessageAskIDs(ctx, msg.ID) + if err != nil { + t.Fatalf("MessageAskIDs: %v", err) + } + if want := []string{"ask-first", "ask-second"}; !reflect.DeepEqual(ids, want) { + t.Fatalf("ask ids = %v, want %v (asks only, in block order)", ids, want) + } + + // An unknown id is a benign empty slice, never a distinct not-found. + unknown, err := s.MessageAskIDs(ctx, MessageID("ghost")) + if err != nil { + t.Fatalf("MessageAskIDs(unknown): %v", err) + } + if len(unknown) != 0 { + t.Fatalf("unknown message ask ids = %v, want empty", unknown) + } +} + // The regression a shared-helper change would cause: AnswerAsk must keep working // through the UN-authorized updateMessageBlocksExec core. AnswerAsk gates on // membership itself (its FOR UPDATE find-and-lock JOINs channel_members) and