feat(runner): SEA-1364 C2 - Gateway telemetry-ingest handlers - #24
feat(runner): SEA-1364 C2 - Gateway telemetry-ingest handlers#24mattwilkinsonn wants to merge 3 commits into
Conversation
| func (s observedConversationSink) PostAgentMessage(_ context.Context, sessionID string, _ *compassv1.MessagePosted, updated *compassv1.MessageUpdated, idempotencyKey string) error { | ||
| kind := "posted" | ||
| if updated != nil { | ||
| kind = "updated" | ||
| } | ||
| 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)) | ||
| slog.String("session_id", sessionID), slog.String("kind", kind), slog.String("idempotency_key", idempotencyKey)) | ||
| return nil | ||
| } |
There was a problem hiding this comment.
When a valid durable conversation frame reaches the production sink, PostAgentMessage only logs that it was observed and returns success without committing or publishing it. The gateway consequently acknowledges the agent and caches the idempotency key even though the message was lost, suppressing retries with that key.
Greptile SummaryThis PR implements the Runner Gateway telemetry-ingest seam.
Confidence Score: 2/5The PR does not appear safe to merge because durable frames can still be acknowledged without being committed, and stalled publisher teardown can block all later telemetry. The production sink still discards conversation frames while returning success, the gateway treats stream Send completion as durable delivery without a per-frame server acknowledgement, and releasePublisher can retain the shared publisher mutex indefinitely while waiting for CloseAndReceive. The earlier sequence-restart teardown issue is fixed by the Gateway-scoped counter, but the remaining previously reported failures still require correction. Files Needing Attention: go/server/sinks.go, go/internal/runner/gateway/post_conversation_frame.go, go/internal/runner/gateway/publisher.go
|
| Filename | Overview |
|---|---|
| go/internal/runner/gateway/post_conversation_frame.go | Adds durable-frame validation, forwarding, retry deduplication, and response handling; the previously reported missing commit acknowledgement remains. |
| go/internal/runner/gateway/publisher.go | Adds shared ordered publishing and preserves sequence numbers across publisher replacement, but the previously reported unbounded teardown lock remains. |
| go/internal/runner/gateway/publish.go | Adds streaming telemetry forwarding and control-ack routing through the shared publisher. |
| go/internal/runnerhub/hub.go | Threads durable idempotency keys through Deliver into the conversation sink. |
| go/server/sinks.go | Extends the production sink signature with the idempotency key while retaining the previously reported no-op write-through. |
| proto/compass/v1/runner.proto | Adds the durable-frame idempotency key to PublishEventsRequest. |
Reviews (4): Last reviewed commit: "fix(compass): unshare the publisher stre..." | Re-trigger Greptile
This stack of pull requests is managed by Graphite. Learn more about stacking. |
88ec220 to
a1b9207
Compare
a1b9207 to
c040720
Compare
Implement RunnerService.CommitConversationFrame, the durable at-most-once counterpart to the loss-tolerant PublishEvents/Deliver path (#24 / OQ-3). The handler resolves the relayed session_id to its bound agent account (fail-closed, mirroring RelayCommsCall: unbound -> CodeNotFound, never a stale account or bootstrap admin), then commits one agent-authored conversation frame at most once keyed on the agent-minted idempotency_key. The key threads to the store's client_request_id, so a retried frame hits AppendMessage's ON CONFLICT (author_account_id, client_request_id) path, writes no second row, does not re-fan MessagePosted, and returns the original row's id -- committed=true on both a fresh commit and a replay. - comms: CommitAgentPostKeyed threads idempotency_key -> ClientRequestId; CommitAgentUpdateKeyed is a documented pass-through (an UPDATE is idempotent by replacement, addressed by message.id, so no dedup key is needed and no store-write signature changes). Both added beside the untouched unkeyed methods the Deliver-path sink still drives. - runnerhub: Hub.CommitConversationFrame (nil-comms -> CodeUnavailable; unbound session -> CodeNotFound) delegates to commitFrame, a oneof dispatcher (posted -> keyed post, updated -> keyed update, neither -> CodeInvalidArgument). Comms errors propagate as-is (already Connect-coded via edgeError) so the Runner's retryable/terminal split reads the right code; a non-commit is never committed=false with a nil error. - handler: (*Handler).CommitConversationFrame with the runnerSubjectFrom guard, fully implementing the generated interface. - seq shipped as 0 (deferred): the downstream consumer reads neither message_id nor seq today; the proto doc-comment carries the caveat. Tests: 5 hub unit tests (fail-closed NotFound, nil-comms Unavailable, neither-variant InvalidArgument, keyed-attribution happy path, error propagation) and 4 comms pgtests including a deterministic at-most-once replay proof (row count stays 1, original id, store head does not advance). Refs SEA-1364 Co-Authored-By: seal <noreply@sealedsecurity.com>
Ports sealed#894 into the standalone repo. Implements Publish and PostConversationFrame on the Runner Gateway, two of the three handler bodies the agent<->Runner seam is missing; Control remains unwritten. The generated files are regenerated in this tree rather than carried as a patch: a generated file is only correct relative to the toolchain that produced it, and sealed's pin differs from this repo's. compass-proto:drift passes on the regenerated output. Stacked on the sun_path cap branch, and the dependency is measured rather than assumed. On main, TestIntegrationProvisionStartRelayToStoreAndBus hangs to a 600s deadline: main has no sunPathMax guard and no shortRuntimeDir helper, both of which live only on that branch, so this port's own integration test cannot pass beneath it. Stacked, the same test passes in 12.8s. The rebase also collided in socket.go, where the length pre-check and this PR's cancel parameter are independent additions to one function; both are kept. Two callsites of listenAgentSocket carried the pre-cancel signature through the rebase with no file-level conflict, plus a skeleton comment quoting the same stale shape. All three updated. Refs SEA-1364 Co-Authored-By: seal <noreply@sealedsecurity.com>
…nnot restart it seq lived on sessionPublisher, so any release-then-reacquire inside one session restarted stamping at 0. runner.proto states runner_seq is monotonic across the Runner's whole event stream, and the hub flags only seq > lastSeq+1, so replayed low seqs are accepted and in-transit loss inside the replayed range stops being detectable. The durable path releases the publisher on any upstream forward failure, which puts this on a common path. Move the counter to the socket-lifetime Gateway as a seqCounter whose mutex is still the allocate-and-send critical section, so allocation order remains emission order and a replacement publisher's first Send cannot interleave with the outgoing publisher's close. Also fixes two test defects found while proving it: fakeSessions recorded without synchronization (latent until a concurrent caller existed), and the wedge regression test left a capturePublish handler parked on a bounded channel at cleanup, hanging httptest.Server.Close and timing out the package in 3 of 8 runs at -race -count=10. Now 0 of 8. Refs SEA-1364 Co-Authored-By: seal <noreply@sealedsecurity.com>
…d sequence Review of the counter hoist found two defects the hoist itself introduced. Sharing the mutex as well as the counter coupled independent publishers. acquirePublisher installs a replacement and closes the stale publisher outside pubMu, so a CloseAndReceive round-trip against an unresponsive-but-connected Server blocked every forward on the replacement's separate upstream stream — with no timeout on that path, since the stream is bound to the socket-lifetime context. Two distinct streams need no mutual ordering: the hub keeps one global high-water mark and cannot observe an interleaving between them. seqCounter now carries only the counter, and each publisher holds its own stream lock across allocate-and-send, so emission order still equals allocation order. forward incremented before the Send and did not roll back on failure. Pre-hoist the counter died with its publisher; now it is socket-lifetime, so a failed forward burned a number permanently, and the hub reads a skipped number as in-transit loss — a durable frame correctly erring back for retry made the Server report a loss that never happened. Adds two deterministic guards, both proven by mutation, neither needing -race: reverting the hoist reddens exactly the two counter tests; dropping the rollback reddens exactly the burn test with the [1 3] hole. The pre-existing concurrent test is demoted to a supplementary check and its detection rate corrected to the measured 26-of-40; the wedge test's hang figure is corrected to the reproducible 6-of-8. Stale scope comments in publisher.go's header and Gateway.seq now state the two-counters-one-high-water-mark hazard precisely. Refs SEA-1364 Co-Authored-By: seal <noreply@sealedsecurity.com>
c040720 to
eec89d9
Compare

Ports
sealedsecurity/sealed#894into the standalone repo. ImplementsPublishandPostConversationFrameon the Runner Gateway — two of the three handler bodies theagent↔Runner seam is missing.
Controlis still unwritten anywhere, so this does notclose the seam on its own.
Stacked on
compass-runner-sea-1440-sunpath-cap(#12), and the dependency is measured, not assumed.Why it must stack on #12
I had this ordered the other way and the measurement reversed it:
mainhas neither thesunPathMaxguard nor theshortRuntimeDirhelper — both liveonly on #12 (
grep -c sunPathMax→ 0 on main, 5 on #12). This PR modifiesintegration_pgtest_test.go, and that test cannot pass beneath #12, so the stack is acorrectness requirement rather than a convenience.
The same 600s hang reproduces on clean
mainwith this port stashed, which is whatestablishes it is not this diff.
What the rebase collided on
socket.goconflicted: #12 adds the path-length pre-check, this PR adds acancelparameter to
listenAgentSocket. Independent additions to one function — both kept.Two callsites carried the pre-
cancelsignature through the rebase with no file-levelconflict, plus a skeleton comment quoting the same stale shape. The compiler reports only
the first, so I enumerated instead:
Generated files are regenerated here, not patched across
The port carries
runner.pb.go,agent.pb.go,agent_gateway.pb.goand two_pb.tsfiles. A generated file is only correct relative to the toolchain that produced it, and
sealed's pin is not guaranteed to be this repo's, so I ran
moon run compass-proto:genin this tree and committed what it emitted.
Base verification
Every one of the 26 files sealed#894 touches was compared between sealed's merge-base
(
fe4e8424) and compassorigin/main:Byte-identical, so the port is a clean apply rather than a hopeful one.
git apply --checkreturned 0 independently.
Gate: run locally, because this repo has no CI
sealedsecurity/compasshas no workflows — aCLEANPR here means nothing ran, noteverything passed. Everything below is local testimony, at
62b2309, with local ==remote == PR head confirmed by
rev-parse.-count=1is deliberate:moon run compass-go:test --forcebypasses moon's cache butnot
go test's per-package cache, and it reported 15 packages(cached)— a greenthat executed nothing.
internal/boardfinishes in milliseconds. That is an in-memory suite, not a skip — the six--- PASSlines are enumerated rather than inferred fromok.The drift gate is proven capable of failing, not merely observed passing: mutating a
generated header to
protoc-gen-go v9.9.9producestask_runner::run_failed; restored andre-verified byte-identical.
moon run :cicannot currently complete in an agent shellNot this PR, and not any diff —
compass-go:fmtfails on cleanorigin/main:The empty file list is the tell.
gofmtresolves to~/.proto/shims/gofmt, which printsa 165-byte NDJSON banner on stdout (
"Detected an AI agent environment"), and the taskis
out=$(gofmt -l .); if [ -n "$out" ]; then … exit 1; fi. The banner is captured as$out. Filtered,gofmt -l .returns 0 filenames on both main and this branch —nothing is unformatted.
It fires only for agents, so a human sees green and all six of us see red on identical
code. Reported to compass-repo; not fixed here, since
go/moon.ymlis not my file andthe obvious patch (
gofmt -w) would rewrite generated files and redden drift.Open Questions — two findings this port carries, unfixed
sealed#894 was
BLOCKEDon two deliberately unresolved review threads.sealedsecurity/compasshas no branch protection (
protection→ 404,rulesets→[]), so nothing here holdsthem. Reproducing them rather than letting the move drop them silently:
1.
PostConversationFrameacks a frame that is never committed.observedConversationSink(
go/server/sinks.go:48-56) logs and returnsnil, so a conversation frame creates no commsrow and no
SubscribeCommsevent while the handler reports success. An ack attests to acommit that did not happen, and a retry short-circuiting on that key makes the frame
permanently absent. Lands as #900 T3 (compass-server's).
2. That sink's stated blocker is stale. Its comment (
sinks.go:37-43) says the realwrite-through "needs the session→channel mapping T5's store adds." That mapping already
landed — verified present on compass
main. Three lanes read past that comment andconcluded the work was gated; it was gated by prose, not by reality. The comment should die
in the same diff as the fix, restated rather than deleted.
Both are verified live on compass
mainas of this PR. Neither is introduced here — thisport makes them reachable, which is why they are stated at the top of the merge decision
rather than left in a sealed thread that did not travel.
RunnerSeq could restart mid-session — found by a new test, fixed here
Reviewing this PR's own durable path surfaced a wire-contract defect that is worse than a
gap, because it is silent.
seqlived onsessionPublisher(publisher.go), so any release-then-reacquire insideone session restarted stamping at 0:
The hub flags a forward jump and nothing else, so replayed low seqs are accepted and a
genuine in-transit loss inside the replayed range stops being detectable. Not a false alarm
— a silenced one.
What made it live rather than theoretical is in this diff: the durable handler releases the
publisher on every upstream forward failure (
post_conversation_frame.go), which it must,or one failed forward wedges every later durable frame for the session. So a single flaky
forward blinded loss detection.
Fixed by hoisting the counter onto the socket-lifetime
Gatewayas aseqCounterwhosemutex remains the allocate-and-send critical section — allocation order still equals
emission order, and a replacement publisher's first
Sendcannot interleave with theoutgoing publisher's close.
relay.go:114-119defers per-Runner (vs per-session) scope to T9. This does not closethat — it makes the sequence unbroken within one Runner link.
Red-green, by mutation
Only
TestReleaseDoesNotRestartTheSequencereddens under that mutation.Note the discriminator: 0/10 without
-race, 10/10 with it. The window needs thescheduler perturbation to be entered at all, which is why no existing test caught it and why
a non-race run is not evidence here.
Two test defects fixed on the way
Both were latent — real, and invisible until something exercised them.
The second was mine, and it was measured rather than reasoned about:
No retry, no skip, no raised timeout — the ordering is now deterministic.
Verification claim for this commit
locally-verified, never CI-verified — this repo has no pipeline (.github/workflows404,
rulesets: []), so a green render here is over an empty required set and means nothing.What I actually ran, in the devenv shell:
Refs SEA-1364
Open Questions — review round (2026-07-29), design forks parked for Matt
A fresh
reviewpass over this diff (correctness + concurrency + security + test-adequacy,one adversarial reviewer) reconfirmed the two sink findings above and surfaced two more that
are design forks, not mechanical fixes — parked here rather than decided by proxy.
3. The durable ack attests to a buffer, not a commit — and the shared-stream design
makes that inherent, so the T5 sink alone does not close OQ-1.
PostConversationFramereturns success (and caches the idempotency key) the instant
pub.forwardreturns.pub.forwardis onlystream.Send(...)on the shared socket-lifetimePublishEventsclient-stream — an HTTP/2 client
Sendthat enqueues subject to flow control. The Server'sterminal per-frame result is produced only by
CloseAndReceive, which the durable pathnever calls per-frame (the stream stays open across many frames). So any server-side loss
after Send-accept (handler crash, sink error, dropped stream) permanently loses a durable
frame while the agent's retry is suppressed. OQ-1 above frames this as "the sink returns
nil"; this is the sharper statement — even once the T5 committing sink lands, acking onSend-accept still violates delivered-or-erred, because "buffered by the Runner" ≠ "committed
by the Server." Fork (proto seam, shared with compass-server): (a) give
PostConversationFrameits own unary upstream RPC returning a per-frame commit ack (cleanest— separates the durable path from the lossy
Publishspine), or (b) add an application-levelper-frame ack on
PublishEventsthe Runner awaits before returning success. The decisiontouches
proto/compass/v1/runner.proto, so it is settled with compass-server, then ruled byMatt.
4.
releasePublisherholdspubMuacross an unboundedCloseAndReceive— a session-wideliveness hazard — and the ordering rationale that justifies it is narrower than the doc
claims.
releasePublisher(publisher.go) holdsg.pubMuacrosspub.close()→stream.CloseAndReceive(), which blocks on the server's terminal response with no per-closetimeout on a
g.baseCtx(socket-lifetime) stream. On an unresponsive-but-connected upstream,every later
Publish/PostConversationFrameblocks on the same mutex → session-widetelemetry outage from one clean stream-end. The lock is held deliberately — to stop a
second publisher opening while the old stream drains (out-of-order-across-streams). But
acquirePublisher's own seqCounter doc (publisher.go:57-66) already states the oppositeprinciple ("never hold
pubMuacrossCloseAndReceivebecause it can block") and closes itsstale publisher outside the lock. The ordering concern is also narrower than stated:
hub.recordSeq(runnerhub/hub.go:227) flags a gap only on a forward jump(
seq > lastSeq+1); a delayed low seq — exactly what a cross-stream reorder produces — issilently accepted,
seenGapstays false. So the reorder the held lock guards against isnot even detected by the hub. Fork (concurrency): bound
CloseAndReceiveunder acontext.WithTimeoutderived frombaseCtx, and/or capturepub+ clearg.pubunderpubMuthen close outside the lock (matching the stale-publisher path), having establishedthe hub tolerates the one-frame reorder. Do not silently keep the unbounded hold. Ruled by
Matt.
5. (medium) No test defends the at-least-once contract. The "delivered-or-erred" tests
exercise only a dead transport (pre-closed server, so
Senditself fails). None exercisesthe real hazard: a
Sendthat is accepted but whose frame the Server then fails toprocess/commit — the exact case OQ-3 describes, under which the unary returns success and
every test stays green. The suite codifies Send-accept as the success criterion, which is
the defect. The fix is contingent on the OQ-3 ruling (the test's shape depends on which ack
model lands), so it is batched with it rather than written now.
These join OQ-1/OQ-2 above as the merge-decision surface. Nothing here is introduced by this
port; the port makes the durable path reachable, which is why the forks are stated at the
top of the merge decision. Tracked: SEA-1364 (this lane); the runner-side residue (OQ-4 lock-across-close +
OQ-5 at-least-once test) as SEA-1545; the sink write-through as compass-server's #900/T3.
Rebased onto post-#32/#42 main + reconciled (2026-07-30)
Reopened per Matt's directive (reopen-in-place, not a fresh PR — keeps number + history) and rebased
c040720→eec89d9onto current main (f5fb433, post-#32 T3 write-through + #42 sink-ack proto). Stack base is nowmaindirectly (the #12 sunpath fix merged).Sink-seam reconcile — went KEYLESS (ruled by compass-server, owner of the write-through): #32 replaced the T4
observedConversationSinkstub with the realcommsConversationSinkwrite-through, whoseConversationSink.PostAgentMessageis account-front + keyless:(ctx, account, sessionID, posted, updated). This PR adopts that signature verbatim across the 5 collision files (hub.go,sinks.go,helpers_test.go,integration_pgtest_test.go,board/deliver_integration_test.go). At-most-once dedup ships via theCommitConversationFrameunary keyed path (compass-server's handler + the sink-ack swap), not this Publish-spine sink — which the swap deletes. Soagent_caller.gois untouched here (byte-identical to #32).RunnerEvent.IdempotencyKey(+PublishEventsRequest.idempotency_key) is retained: the Runner gateway carries the key to the sink boundary (32 gateway tests assert it rides the envelope), but the hub does not thread it into the keyless sink — documented at the field. Whether to drop the now-hub-unconsumed proto field is a follow-up tied to the swap landing, not decided here.OQ-1/OQ-3 (the
observedConversationSinkacks-without-committing findings) are RESOLVED by this reconcile: the stub is gone;commsConversationSinkcommits durable rows viaCommitAgentPost/CommitAgentUpdate. The delivered-or-erred hardening (OQ-3 sharpened) ships in the sink-ack swap (SEA-1561) on top of this.Verification (devenv shell, head
eec89d9):go build ./...clean,go vet ./...clean,go test -race -count=1 ./internal/runnerhub/... ./internal/board/... ./internal/runner/... ./server/...all ok, fullmoon cigate green through thehkpre-push hook (lint/drift/breaking/vuln/licenses + tests). One contention flake on first push (GOARCH-NDJSON-bleed →compass-go:lintpackage-load corruption + cascaded nilaway) cleared deterministically on a warm-cache serial retry (compass-go:lint= 0 issues).#38 (C3 Control) restacked onto
eec89d9(cee7015→3852944), pushed green.