feat(runner): C3 runner control producer (AgentGateway.Control)

[sealedsecurity-bot]

Summary

The Runner's control producer: AgentGateway.Control plus the ControlSender seam the
session lifecycle writes through. Stamps a monotonic control_seq, retains each op past
the agent's ack cursor, and drains it to whichever subscription is currently bound.

Ports sealedsecurity/sealed#911 (SEA-1364 C3) into sealedsecurity/compass. The port is
faithful — control.go is byte-identical to the sealed source (both trees share the
github.com/sealedsecurity/compass/go module path, so no import rewrite was needed), and
the three integration edits (gateway.go, socket.go, host.go) apply against C2's tree
with only line-offset shifts.

Stacked on #24 (C2)

C2 (#24) defines ControlRouter — the ack-routing seam with a noopControlRouter default
and a SetControlRouter injection point, built explicitly for this lane to fill. This PR:

• Serve injects the real *controlProducer, so the session lifecycle can reach the agent;
  before this the producer was constructed nowhere in production and only tests could reach it.
• Control upgrades the injected router to a controlLane (C2's interface plus the drain);
  an ack-only router refuses the subscription (CodeUnimplemented) rather than accepting a
  stream nothing can ever write to.
• SocketListener.Bind/RetireSession thread through agentHost.Start/Stop under the same
  lock that records/drops the session, so control state is created and reclaimed with the
  session on a socket that outlives it. C2's seam and its test fake are untouched.

Base: compass-runner-1364-c2-telemetry-ingest. Merge #24 first.

Three invariants carry the contract

• Ordering — one per-session queue drained by one goroutine, so Send order is delivery
  order.
• Retention — Send success means "durably queued until acked", not "handed to a socket".
  Ops live past the agent's ControlAck cursor, so an agent or container replacement loses
  nothing and needs no durable store of its own. Bounded at maxRetainedOps; past the cap
  Send returns CodeResourceExhausted (backpressure, an ack frees room) rather than evicting
  an op already reported durably-queued.
• Replay barrier — live ops are held while a replay is in flight and released on the
  agent's ReplayCompleteAck (routed here from the Publish stream via C2's seam). Replay-path
  ops are exempt from both the hold and the cap: the barrier's own release is the one op whose
  ack lifts it, so holding or rejecting it would wedge the stream permanently.

Untrusted-side hardening (agent drives acks + subscribe)

• AckControl clamps the ack cursor to the highest issued seq — an unclamped cursor past
  nextSeq would make every future op invisible to every future subscription.
• applied_above is agent-sized and unordered; boundedAboveSet intersects it against live
  retention (not the seq window, which is unbounded over a session's life), capping what one
  ack can allocate for any input in any order.
• Agent-driven paths (serve, AckControl, ReleaseReplayBarrier) resolve sessions through
  existingSession (non-creating): an ack or subscribe racing Stop cannot resurrect an
  entry nothing would ever retire again.

Concurrency scopes, deliberate and documented

• sink.Send runs outside the session lock (it is network I/O): holding the mutex across
  it would stall every Send and ack behind one slow socket. The cost — an ack landing mid-batch
  can re-deliver one already-acked op — is covered by the at-least-once seq-dedup and is
  documented so it is not "fixed" into a lock-across-I/O bug.
• Wake channels are per-subscription, closed on rebind, so a displaced drainer cannot
  absorb the live one's wake token (the stolen-wakeup strand).

Tests

Four new test files (control_test.go, control_untrusted_test.go, retire_wiring_test.go,
e2e_retire_test.go), covering: seq stamping + ordering, pointer-reuse seq distinctness,
retain-with-no-subscription, takeover transfer + the two-stale-drainer wake, replay_complete
through the barrier, SendIfLive liveness, the retention cap, untrusted ack clamping +
applied_above bounding, and the Bind/Retire wiring (both the leak assertion and the
nil-producer guards on Start/Stop).

Verified locally (go 1.26.5, devenv):

• go build ./... clean; go vet -tags "pgtest unix" ./internal/runner/... clean.
• go test -tags "pgtest unix" -race: internal/runner ok, internal/runner/gateway ok,
  and downstream internal/runnerhub + internal/runtime ok — no regression on the stack.
• Pushed through the dev shell, so the hk pre-push gate (moon run :ci) ran: 23 tasks
  completed.

Refs SEA-1364
Ports sealedsecurity/sealed#911

Co-Authored-By: seal noreply@sealedsecurity.com

Open Questions (parked for Matt — '/home/mattw/.agents/rules/decision-authority.md')

OQ-1 — serve()-vs-Retire() TOCTOU (SEA-1550, review floor: medium). Greptile (P1) and
the mandatory review agent (graded medium) both flagged a race in control.go: serve
resolves the session under p.mu, releases it, then acquires s.mu ~12 lines later to bind —
a full Retire fits between, so serve binds on a detached session and the drainer parks on a
wake nothing will re-close. Benign today (Retire is Stop-driven, Stop kills the agent child
→ the drainer's Control-RPC ctx cancels promptly, a complete backstop), but a live goroutine +
session-state leak the moment T9 in-process reattach lands (host.go:194), if reattach ever
retires without dropping the agent connection. Hard blocker on T9.

Carried verbatim from sealed#911 (control.go is byte-identical), so NOT port-introduced.
Two paths, and this is Matt's call because it is a control-flow + source-of-truth fork:

• Fix here: hold p.mu across the bind block in serve (preserves the existing
  p.mu→s.mu order Retire uses, so no deadlock; closes the window). Diverges compass's copy
  from byte-identical sealed.
• Fix upstream first: keep compass byte-identical to sealed#911, fix in sealed, re-port.

Not auto-fixed: a lock-ordering change to concurrency control flow that also forks the
source-of-truth copy is a design decision, not a mechanical fix.

[linear-code]

SEA-1364

[greptile-apps]

Greptile Summary

Adds the Runner control producer and integrates it with gateway and session lifecycle handling.

• Implements sequence stamping, retained delivery, acknowledgements, replay barriers, subscription takeover, and bounded retention.
• Wires producer creation into gateway serving and session binding/retirement into host lifecycle operations.
• Adds unit, concurrency, untrusted-input, wiring, and end-to-end retirement tests.

Confidence Score: 3/5

The PR is not yet safe to merge because retirement can still race subscription binding and leave a detached Control stream alive.

serve releases the producer lock after resolving a session and only later acquires the session lock to bind; Retire can delete and tear down that session between those operations, after which serve installs fresh live subscription state on the detached object with no later retirement to close it.

Files Needing Attention: go/internal/runner/gateway/control.go

Important Files Changed

Filename	Overview
go/internal/runner/gateway/control.go	Implements control production and subscription draining, but the previously reported retirement-versus-binding race remains at current HEAD.
go/internal/runner/gateway/gateway.go	Wires the control producer into the served gateway.
go/internal/runner/gateway/socket.go	Threads control-state binding and retirement through the socket listener lifecycle.
go/internal/runner/host.go	Invokes control binding and retirement from session Start and Stop.
go/internal/runner/gateway/control_test.go	Adds broad behavioral and concurrency coverage for control sequencing, retention, takeover, replay, and retirement.
go/internal/runner/gateway/control_untrusted_test.go	Covers hostile acknowledgement values and post-retirement agent traffic.
go/internal/runner/gateway/retire_wiring_test.go	Verifies socket lifecycle calls reach the control producer.
go/internal/runner/e2e_retire_test.go	Verifies ordinary Stop-driven retirement terminates the active Control subscription.

_{Reviews (2): Last reviewed commit: "feat(runner): C3 runner control producer..." | Re-trigger Greptile}