Skip to content

M2-07: Agent consumers over the gateway and REST

Status: ready | Depends on: M1-65 | Milestone: M2

Goal

After this ticket an agent adapter reads the event log as a named consumer with a durable cursor, over the gateway WebSocket and over REST, exactly as 04 "Agent adapters" ("Connection and cursor", "Ack and redelivery") and 09 "Connection and cursor" require. Each pair of agent and consumer name owns one row in agent_consumer_cursors. The first connection of a consumer starts at head_seq, every later one at acked_seq + 1; at most agent.max_inflight_events unacknowledged events are on the way, then the server waits; an event that is not acknowledged within agent.ack_timeout comes again with the delivery field redelivered: true; acked_seq is saved at the latest after agent.ack_persist_interval and at shutdown. The newest consumer wins (R-SY-09): a second identify with the same consumer takes it over and closes the older connection with 4015, and the fenced save of 03 keeps that rule true when the takeover happens in another server instance. The REST operations getConsumerState, pollAgentEvents and ackAgentEvents give adapters without a WebSocket the same stream with the same rules. Without this an adapter either misses events or works twice, and the invocation tokens of M2-08 have no reliable stream to travel on.

Read first

  • docs/PLAN.md: section 3 "Way of working per ticket"; section 8 "M2 Agent platform, extensions and terminal" (Goal, Scope, Out of scope: "fencing across several nodes (M7)", Exit criteria items 1 and 10, Test strategy: "restart of the server with open acks", Red team, Risks), block "agents and extensions" with the rows M2-06, M2-07, M2-08, M2-09, M2-17, M2-20, M2-26, M2-29, M2-31, M2-36; section 7 rows M1-51, M1-63, M1-64, M1-65, M1-68; section 13 rows M7-04, M7-08, M7-17.
  • Implied dependencies (in the closure of M1-65, therefore not in the header): M1-16 (event log, envelope, DeliveryFields), M1-19 (rate limits), M1-20 (idempotency), M1-25 (action registry), M1-28 (authorize), M1-29 (authz.Admin, access.changed), M1-44 (API tokens), M1-45 (revocation), M1-51 (agents), M1-56 (messages), M1-63 (catch-up), M1-64 (gateway connection).
  • docs/spec/04-sync-protocol.md: 15 "Agent adapters" with 15.1 "Connection and cursor" (points 1 to 5), 15.2 "Ack and redelivery" (points 1 to 4), 15.3 "Invocation and run start" step 2 (the delivery check that M2-08 hooks in), 15.6 "Webhook as transport" (out of scope); 2.2 "Envelope" (delivery fields redelivered, invocation_token, trace_context); 2.3 "Frames on the WebSocket"; 6.2 "Ops" (ack, error keeps the connection); 6.3 "Lifecycle" (identify for agents with consumer, default default); 6.4 "Close codes" (4002, 4015); 7.1 "Cursor"; 7.2 "Flow"; 7.3 "resync_required"; 7.4 "REST variant"; 7.5 "Sync epoch" point 4; 8.1 "Basic rule"; 9 "Delivery guarantee"; 10 "Ordering"; 17 "Decisions of this chapter" (R-SY-02, R-SY-03, R-SY-09, R-SY-17, R-SY-23); 18 "Configuration values of this chapter" (agent.ack_timeout, agent.ack_persist_interval, agent.max_inflight_events).
  • docs/spec/03-data-model.md: table agent_consumer_cursors (all columns, PK, FK, the fenced save below the table); table agents; 3 "Retention classes" (security); 6.3 "Lock order"; 6.6 "Notifying the gateway processes"; 6.7 "Sync epoch and restore".
  • docs/spec/09-agents.md: 2.1 "Records" (row "Read position"); 4.2 "Transports"; 4.3 "Connection and cursor" (acceptance before ack, duplicate detection); 4.4 "Events to the adapter"; 4.5 "Invocation token" paragraph "Check in the adapter" (an event without accepted token is not acknowledged and comes again after agent.ack_timeout) and "Keys and rotation" point 3; 15 "Reliability".
  • docs/spec/05-actions.md: rules 1 to 4 of 1 "Binding rules"; 2.1 "Column "Permission"" (agent-self); 2.7 "Effect (effect)"; 3 "Rate-limit classes" (rl.agent_runtime); row agent.events.consume in 4.10 "Agents, commands, runs, interactions"; 6 "Durable event types and receive action"; 7 "WebSocket ops and action" (row ack); R-AZ-01, R-AZ-04.
  • docs/spec/02-architecture.md: 4.1 "Rules" (rules 1, 4, 6); 4.3 "Ports, events and sources per module" (rows agents: Consumers; gateway); 4.4 "Allowed dependencies" (agents may use only identity, authz, chat; gateway may use agents and sync); 4.6 "Actions per module" (agent.events.consume belongs to agents); 4.7 "Event types and WebSocket ops per module" (ack belongs to agents); 5 "Layers within a module" (rule 3); 7.3 "Other checks in the gate" (row "Routes and ops": the handler lies in the module that owns the action); 10 "Gateway"; 11.2 "Log tail per api process"; 11.3 "Permission cache"; 11.4 "Hub and delivery"; 11.7 "Cost of serialization"; 14.4 "Failure of a dependency"; 15.3 "Shutdown" (role api, step 3); 16 "Scaling path" (bullet "Agent consumers").
  • docs/spec/06-api.md: 2 "Basic rules" (rows "Self reference", "No effect", "Unknown values"); 5 "Errors" (points 3, 4, 6, 7); 6 "Pagination" (points 3 and 5); 7 "Idempotency" (generated table: ackAgentEvents is exempt, "agent runtime: monotonic or compare-and-swap"); 9 "Rate limits"; 11 "Real-time gateway" (points 4 and 5); 16 "Endpoints" (rows getConsumerState, pollAgentEvents, ackAgentEvents); 17 "WebSocket ops and events" (row ack).
  • docs/spec/07-permissions.md: 8.1 "Delivery filter" (row "Agent consumers"); 11.2 "Required tests" (AZ-P1, AZ-P2, AZ-P9, AZ-P13).
  • docs/spec/20-privacy.md: 2 "Data categories per table" (row agent_consumer_cursors: not personal, class security, deleted with the agent).
  • docs/api/openapi.yaml: operations getConsumerState, pollAgentEvents, ackAgentEvents, listEvents (for comparison); parameter ConsumerPath; schemas ConsumerState, ConsumerAck, EventPage, EventEnvelope (member redelivered), GatewayIdentify (member consumer), GatewayResume, GatewayAck, GatewayCursor, GatewayCatchupComplete, GatewayResyncRequired, GatewayError; response Gone; x-problem-codes of Problem (validation_failed, forbidden, not_found, resync_required, rate_limited).
  • docs/api/asyncapi.yaml: channel gateway (description: one connection per agent consumer, 4015), operation receive.ack, message op.ack, x-close-codes entry 4015 consumer_replaced.
  • ADRs: 0007 (migrations), 0008 (hexagonal modules), 0013 (spec first), 0017 (one authorization function, 404 before 403), 0022 (agents outside the server process), 0050 (module register), 0052 (one event envelope, delivery fields), 0053 (no JSON null), 0059 (sync epoch), 0065 (maintenance mode, point 4: ack stays allowed for agents with a running run; the holder column).
  • docs/DECISIONS.md: D-K-A3 (fencing column), D-W-100; section 10 for the format of new rows.
  • docs/security/threat-model.md: 5.5 "Gateway (WebSocket)", 5.10 "Agent adapters and runtime"; register rows T-TK-02 (stolen agent token), T-AZ-03 (delivery after revocation), T-DS-01 (slow consumer).
  • docs/RISKS.md: RK-37 (several api nodes). docs/OPEN-QUESTIONS.md: OQ-07 (default: multi-node operation in M7, "The code stays node-neutral from M1 on").
  • docs/observability/signals.md 4.3 row bothy.agent.consumer.lag and docs/observability/alerting.md row BothyAgentConsumerLag: read to know they exist; this ticket does not build them (see Out of scope).
  • docs/operations/configuration.md: rows agent.ack_timeout, agent.ack_persist_interval, agent.max_inflight_events, sync.catchup_page_size, sync.cursor_advance_every, sync.resume_window, database.listen_reconnect_backoff_cap, cluster.mode.
  • M1 briefs this ticket builds on: M1-16 (sync.Event, sync.Envelope, sync.DeliveryFields{IncludeTraceContext, InvocationToken, Redelivered}, sync.Tail), M1-25 (authorize.Registry, module Register functions), M1-28 (AgentSelf, visibility of kind agent, authorize.Filter), M1-29 (authz.Admin.CreateGrant, RevokeGrant), M1-44 (agent API tokens with dpop_jkt), M1-45 (agent.disable and token revocation close with 4012), M1-51 (agents port Agents, internal/agents/adapters/httpapi), M1-63 (CatchUp.Check, CatchUp.Page, envelope encoding, trace_context for agents), M1-64 (session store, Delivery{Register, Attach, Detach}, op dispatcher, gateway_ops in api/operations.yaml, close code constants, wstest client), M1-65 (hub, hub position, authorize.Delivery(ctx), Observe, catch-up runner, queue purge, shutdown step with 4013, two server instances in one test binary).
  • Merged code on branch main (read with git show main:<path>, no checkout): services/server/architecture/modules.yaml (module agents: table agent_consumer_cursors, action agent.events.consume, ws_ops: [ack]; module gateway: depends_on with agents and sync), services/server/internal/agents/README.md, services/server/internal/gateway/README.md (shutdown: save agent cursors first), services/server/internal/wire/configkeys_messaging.go and configchapters.go (pattern for registering configuration keys).
  • Code: internal/agents (M1-51), internal/gateway (M1-64, M1-65), internal/sync (M1-16, M1-63, M1-65), internal/authz (M1-28, M1-29), internal/kernel/authorize, internal/kernel/actor, internal/kernel/clock, internal/platform/pg/pgtest, internal/identity/adapters/identitytest, internal/wire, services/server/api/operations.yaml.

Fixed requirements that apply

Owner's fixed build rules (apply verbatim):

  1. No MVP. New from scratch and directly in the final state; stages are only the order.
  2. Strictly modular: clear module and service boundaries, dependencies only through defined interfaces, no cross access.
  3. Documentation in code: every public module, class, function and interface with a docstring/doc comment (purpose, parameters, return value, errors, example where useful); comments explain the why, not the what. API documentation is generated from code (OpenAPI from the routes, dartdoc, Python API docs in MkDocs).
  4. Documentation in the repo: README per module/service, ADR for every architecture decision, CHANGELOG per release, keep the documentation site current. A change without matching documentation is not done.
  5. Quality gates (SonarQube, linters, type checking, security scanners, test coverage) block the merge.

Rules for every brief:

  1. Test first: write the failing test, run it and keep the red output for the report, then make it green, then clean up.
  2. No en dash and no em dash in any file you write; the text-rules pre-commit hook rejects them. Use a hyphen or rephrase.
  3. White label: examples and test data use example.com and the people Alex, Sam and Robin; no company, host or personal names, no paths under a home directory.
  4. Limits, windows and thresholds only through named configuration keys listed in docs/operations/configuration.md; code comments and docs name the key, never the number.
  5. Specification first: a change to behaviour, API, event, permission or configuration updates docs/spec/ or docs/api/ in the same change. The core contract (docs/spec/00-glossary.md, 03-data-model.md, 04-sync-protocol.md, 05-actions.md) is not edited by this ticket.

Product requirements implemented here:

  1. One durable cursor per agent and consumer name in agent_consumer_cursors, exactly as 03 defines the table (columns, PK (agent_id, consumer_name), FK to agents(principal_id) with ON DELETE CASCADE, nullable holder_session_id, retention class security); only the module agents reads and writes it (03 table, 02 4.1 rule 1, 20 row agent_consumer_cursors).
  2. The first connection of a consumer sets the cursor to the current head_seq; an adapter never receives events from before its first connection (04 15.1 point 2; 09 4.3).
  3. Delivery runs from acked_seq + 1 in seq order, filtered like people by the agent's own grants and the operator rule, plus personal events with the agent as audience and agent.updated about itself; every delivery, also a redelivery, is admitted with the permissions at delivery time (04 15.1 point 4, 8.1, R-SY-02; 07 8.1 row "Agent consumers").
  4. ack {seq} confirms every number up to seq; the gateway op and ackAgentEvents both use action agent.events.consume (04 15.2 point 1, 6.2; 05 7 row ack; 00 "Ack").
  5. At most agent.max_inflight_events unacknowledged events are on the way; then the server waits (04 15.2 point 2).
  6. Without an acknowledgement within agent.ack_timeout the server delivers again from acked_seq + 1 with the delivery field redelivered: true; delivery fields are set per delivery and never stored in the log (04 15.2 point 3, 2.2, R-SY-17, ADR 0052).
  7. acked_seq is stored at the latest after agent.ack_persist_interval (04 15.2 point 4) and immediately in step 3 of the shutdown of role api, before all connections close with 4013 (02 15.3).
  8. One connection per consumer, the newest wins: a new identify writes its gateway session into holder_session_id and closes the older connection with 4015; acked_seq is saved only with WHERE holder_session_id = <own session>, and a save that finds no row closes the connection with 4015 (04 15.1 point 3; 03 table agent_consumer_cursors; R-SY-09, R-SY-23; D-K-A3).
  9. agent.events.consume is agent-self, class rl.agent_runtime, Audit no, Permissions no; denials of requests and ops are audited by authz, delivery filters write no audit entry (05 4.10, R-AZ-04; 07 AZ-P13). The action and the op belong to agents, so their handlers live there (02 4.6, 4.7, 7.3).
  10. REST: getConsumerState reads the cursor, pollAgentEvents delivers from acked_seq + 1 with at most agent.max_inflight_events per call and redelivery after agent.ack_timeout, ackAgentEvents is idempotency-exempt and monotone (openapi descriptions; 06 7 "Idempotency"; 09 4.2 "Transports"; 04 7.4).
  11. No JSON null in any frame or body (ADR 0053); every frame and body conforms to its schema in docs/api/openapi.yaml (06 15, G1 to G4).

Scope

All paths under services/server/ unless noted.

Create:

  • migrations/<next free five-digit number>_agent_consumer_cursors.sql: table agent_consumer_cursors exactly as 03 (all six columns with types, nullability and defaults, PK, FK with cascade); a comment names the retention class security. Down migration drops the table.
  • internal/agents/consumers.go (port additions, doc comment on each symbol):
    • type ConsumerKey struct { AgentID ids.ID; Consumer string }, type ConsumerState struct { Consumer string; AckedSeq int64; Created bool }, sentinel ErrConsumerReplaced (an errs value of kind conflict).
    • type Consumers interface, implemented by agents/app: Take(ctx, consumer string, holder ids.ID, head int64) (ConsumerState, error) (authorize, create at head when absent, write holder); AcceptAck(ctx, consumer string, seq, head int64) error (authorize and validate, no write); Save(ctx, key ConsumerKey, holder ids.ID, acked int64) error (fenced, monotone); Restart(ctx, key ConsumerKey, holder ids.ID, seq int64) error (fenced, sets the value, used only for resync). The actor comes from the context for Take and AcceptAck; Save and Restart take the key because the gateway's save ticker has no request.
    • Declared source port type ConsumerFeed interface (02 4.1 rule 4: agents declares, gateway implements): Poll(ctx, consumer string, limit int) (FeedPage, error), Acked(ctx, key ConsumerKey, acked int64), Head(ctx) (head int64, epoch ids.ID, err error); type FeedPage struct { Events []json.RawMessage; NextCursor, HeadSeq int64; SyncEpoch ids.ID; Resync string } (Resync empty, retention or cursor_ahead).
    • In Register(r *authorize.Registry) of the port (add the function if M1-51 did not): register the op ack for module agents (M1-25).
  • internal/agents/domain/consumer.go: ValidConsumerName(string) bool (pattern of ConsumerPath), DefaultConsumer = "default", CheckAck(seq, acked, head int64) (advance bool, err error) (Design note 8).
  • internal/agents/app/consumers.go: the Consumers use cases and the three REST use cases ConsumerState, PollEvents, AckEvents (Design notes 9 and 10).
  • internal/agents/adapters/postgres/queries/consumer_cursors.sql with the sqlc queries TakeConsumer (insert at head or set holder_session_id, returning acked_seq and whether the row was created), SaveConsumerAck and RestartConsumer (both with holder_session_id = $holder), AckConsumer (monotone, no holder condition, returning acked_seq), GetConsumer; the repository methods.
  • internal/agents/adapters/httpapi/consumers.go: the strict-server methods GetConsumerState, PollAgentEvents, AckAgentEvents.
  • internal/gateway/domain/consumerwindow.go: the pure delivery window of Design note 5 (no I/O, time passed in).
  • internal/gateway/app/consumers.go: the consumer engine: a runner per consumer connection, the per-process holder index, REST windows, the save ticker, flush on takeover, detach, close and shutdown; internal/gateway/app/ack.go: the handler of op ack; outgoing port DeliveryHook (Design note 7).
  • internal/gateway/adapters/consumerfeed/: implements agents.ConsumerFeed on the engine.
  • internal/gateway/adapters/passdelivery/: the DeliveryHook binding until M2-08; it returns every envelope unchanged; its doc comment names M2-08 as the ticket that binds runs.Delivery.
  • internal/wire/configkeys_agents.go: the keys agent.ack_timeout (duration), agent.ack_persist_interval (duration) and agent.max_inflight_events (int, minimum 1) with chapter 04-sync-protocol.md, following configkeys_messaging.go; reuse the chapter constant if an earlier ticket added it.

Change:

  • internal/gateway (M1-64, M1-65): identify and resume of agent principals (Design notes 3 and 4); Delivery.Attach starts the consumer runner for agent sessions instead of the catch-up runner; the hub keeps agent sessions out of durable dispatch and wakes their runners instead (Design note 6); queue purge also releases the purged sequence numbers from the consumer window; the shutdown step of M1-65 flushes all consumer cursors before CloseAll(4013).
  • internal/sync port, only where the merged code lacks it: a page variant that returns sync.Envelope values (or sync.Event rows) instead of encoded JSON, so the gateway sets DeliveryFields.Redelivered per delivery without parsing JSON; an accessor for the hub position if M1-65 has none (Design note 6).
  • internal/wire: bind agents.Consumers into the gateway engine and consumerfeed into the agents use cases (they depend on each other at run time: use a late setter or a small forwarding type in wire); register the ack op handler with the dispatcher; bind passdelivery; register the save ticker and the REST window sweeper with the lifecycle (M1-11).
  • api/operations.yaml: move getConsumerState, pollAgentEvents, ackAgentEvents from later to enabled (sorted) and ack from gateway_ops.later to gateway_ops.enabled; rerun the route generator (make codegen-check shows no diff afterwards).
  • docs/api/openapi.yaml (spec first): pattern: ^[a-z0-9][a-z0-9_-]*$ on GatewayIdentify.consumer (the pattern of ConsumerPath); minimum: 0 on ConsumerAck.seq and GatewayAck.seq; on pollAgentEvents the response '410': $ref: '#/components/responses/Gone' and a description of the rules of Design note 10 (takeover, first poll at head_seq, window, 410 resync_required); descriptions of getConsumerState (404 for a consumer that never connected) and ackAgentEvents (monotone, 400 above head_seq, 404 unknown). Then make api-lint, make api-tables, make api-checks.
  • docs/spec/09-agents.md 4.3 "Connection and cursor": a paragraph "REST fetch" with the rules of Design notes 8 to 10 and the consumer name pattern; 4.2 row "REST fetch" points to it. English, like the chapter.
  • docs/spec/02-architecture.md: 4.3 row agents "declares ConsumerFeed", row gateway "Implements sources: agents.ConsumerFeed"; 10 "Gateway" and 11.4 "Hub and delivery": one sentence each that agent consumer connections read the log through their own runner up to the hub position and are woken by the hub.
  • docs/operations/configuration.md: regenerate section 4 with make config-reference; make config-doc-check green.
  • READMEs of agents and gateway (English, like the existing files); services/server/CHANGELOG.md under "Unreleased"; docs/DECISIONS.md section 10 rows D-M2-07-1 and following (Design note 15).

No new top-level Makefile target.

Out of scope

  • Fresh invocation tokens, agent.invoke, I2, SR2 and SR4 before each delivery of agent.invocation, agent.followup, interaction.created, command.invoked, and binding runs.Delivery into DeliveryHook (M2-08); signing the tokens (M2-06).
  • Heartbeat over the gateway (heartbeat.agent) and sendHeartbeat, presence, agent.presence_changed (M2-20).
  • The ephemeral command.autocomplete_requested to an active consumer connection (M2-17); run.update and command.autocomplete_result ops (M2-10, M2-17).
  • Webhook as transport for agents (04 15.6, webhook_subscriptions.agent_id): no owner in the plan yet; reported as a plan gap, do not build it here.
  • The fast cross-node signal of a takeover over NOTIFY ephemeral, cluster.mode = multi and the acceptance test with two api nodes (M7-17). The fenced save itself is part of the table's contract in 03 and is built here.
  • Resetting agent cursors in bothy restore finalize (M7-04); the maintenance-mode exception for ack (M7-08, ADR 0065).
  • The metric bothy.agent.consumer.lag and the alert BothyAgentConsumerLag: no ticket of the plan owns them yet, and the metric infrastructure of M1-68 is not in this ticket's dependency closure; do not register the instrument. Name it in the report.
  • SDK handling of consumers and acknowledgements (M2-29, M2-30), the conformance suite (M2-31), load tests on the agent path (M2-36).
  • Changes to the core contract.

Boundaries

  • A new migration needs its line in migrations.sha256: run make migrations-lock after creating the file and before make check; never edit a merged migration (M1-01, ADR 0007).
  • Work in a dedicated git worktree on the branch feat/M2-07-agent-consumers; one writer per worktree.
  • Conventional Commits with the module as scope (for example feat(agents): ..., feat(gateway): ...) and a DCO sign-off on every commit (git commit -s); squash into meaningful commits before the merge (docs/quality/push-discipline.md).
  • Never push to main, never tag and never create remote repositories; the only remote steps are the branch push and the pull request of AGENTS.md.
  • Do not modify files outside the scope above, except the CHANGELOG, README and docs/DECISIONS.md entries named in this brief.
  • Follow the host rules supplied by the launcher (ports, machine rules).
  • Tests bind only 127.0.0.1 and port 0; PostgreSQL runs only as a throwaway container on a random port through pgtest (P-09, P-10). Without a Docker-compatible runtime the integration tests fail; they never skip.
  • Stop only processes you started, by PID. Temporary directories only via mktemp -d, removed at the end.
  • Never read or print secrets. No network calls except package downloads for pinned tools and Go modules.
  • No agent runtime, adapter, SDK or model is started: the agent side in tests is the wstest client of M1-64 or plain HTTP with the agent's API token and DPoP proofs from identitytest; no real third-party service.
  • Time-dependent tests set short values for agent.ack_timeout, agent.ack_persist_interval and sync.resume_window through the configuration loader or use the kernel/clock test clock; no test depends on a default value.
  • Never lower a threshold, disable a gate or delete a test.
  • If the core contract lacks something this ticket needs, stop with Verdict: BLOCKED <reason> instead of building a deviation (P-11).
  • This brief was written before milestone M1 closed (D-W-100). Where the merged code of an M1 ticket names a package, type, function, file or make target differently from this brief, follow the code and list each difference in the report; where it behaves differently from the specification, stop with Verdict: BLOCKED <reason>.

Design notes

  1. Placement (02 4.4, 4.6, 7.3). agent_consumer_cursors, agent.events.consume and the op ack belong to agents, so the cursor use cases, the three REST handlers and the registration of ack live there. Delivery belongs to the gateway (02 10: agent connections are consumers; invocation events go through runs.Delivery before sending), and agents may not use sync or gateway. Therefore one delivery engine in gateway serves both transports: the WebSocket directly and REST through the source port agents.ConsumerFeed, which agents declares and gateway implements (02 4.1 rule 4). The gateway reaches the table only through agents.Consumers. Ruling for docs/DECISIONS.md: one engine for both transports, so the window rules and the later hook of M2-08 exist once.
  2. Table and fencing. The table is built exactly as 03, with holder_session_id. Every takeover writes the holder and every save of acked_seq is UPDATE ... SET acked_seq = GREATEST(acked_seq, $acked), updated_at = now() WHERE agent_id = $1 AND consumer_name = $2 AND holder_session_id = $holder; no row means another holder took over: ErrConsumerReplaced. This holds in both values of cluster.mode, because 03 states the conditional save without restriction and OQ-07 keeps the code node-neutral. Ruling for docs/DECISIONS.md: fencing is always on; M7-17 adds only the faster signal between nodes. These statements take no lock beyond their row and append no events, so they take neither event_seq nor NOTIFY (02 11.7, 03 6.3).
  3. Consumer name. An agent's identify carries consumer, default default (04 6.3); a resume keeps the consumer of the resumed session (GatewayResume has no such member). A value that does not match the pattern of ConsumerPath closes the connection with 4002 (M1-64 note 3, frame does not match its schema; this ticket adds the pattern to GatewayIdentify.consumer). In REST the path parameter is checked by the chain: 400 validation_failed with pointer #path/consumer. Consumers exist only for principals of kind agent (04 6.2 row ack, 6.4 row 4015: "agents only"); for people and apps the member consumer is accepted and ignored, and their delivery stays as M1-65 built it. Ruling for docs/DECISIONS.md.
  4. Agent sessions on the gateway. For an agent the durable cursor decides where delivery starts (04 15.1 point 4), so the gateway ignores cursor and sync_epoch of identify and resume and skips the client cursor check of M1-64 note 12 for agent sessions. Order after ready (the hub registration before reading head_seq stays as M1-64 note 11 and M1-65 built it), in Delivery.Attach, under a per-process lock per ConsumerKey: a. If this process holds the key in another gateway session, save that session's acknowledgement with Save (it still holds the row). b. Take(consumer, holder = this gateway_session_id, head = ready.head_seq): authorizes agent.events.consume; creates the row with acked_seq = head_seq if absent (the first connection starts at head_seq, 04 15.1 point 2); otherwise writes the holder and keeps acked_seq. forbidden closes with 4018, unauthenticated with 4012, a database failure with 4000 (04 6.4 table). c. Close the older connection of step a with 4015, reason consumer_replaced, and remove its gateway session from the store, so a later resume of it gets 4006. Drop a REST window of the key in this process. Log one INFO record with agent id, consumer and both session ids; no audit entry (05: Audit no). d. Start the runner with acked = acked_seq (note 5). A resume keeps the gateway session id and therefore the holder; it restarts the runner from the session's in-memory acknowledgement. If the resumed session's old connection is still open, it closes with 4015 (another process took over the same consumer), not 4017, because agent connections have no device (M1-64 note 15). Ruling for docs/DECISIONS.md: steps a to d, the session end after 4015, and the ignored client cursor.
  5. Delivery window (gateway/domain, pure). State: acked (highest acknowledged), read (highest seq checked by this window), inflight (dispatched durable events with seq > acked, each with the time it was queued), redeliverUpTo, highestSent, the phase (backlog or live) and the count of skipped numbers since the last frame that carried a seq.
    • Budget: agent.max_inflight_events minus the number of inflight entries; at zero the runner reads nothing and waits (04 15.2 point 2).
    • Reading: a page after read up to the bound of note 6. Of the admitted events at most the budget is dispatched; if the page was cut, read becomes the seq of the last dispatched event, otherwise the last checked seq of the page.
    • Ack: seq > acked sets acked, drops inflight entries up to seq and raises read to acked if it was lower; acknowledging is cumulative (04 15.2 point 1).
    • Timeout: the deadline is the queue time of the oldest inflight entry plus agent.ack_timeout. When it passes, redeliverUpTo = max(redeliverUpTo, highestSent), read = acked, inflight is emptied, and reading restarts from acked + 1 (04 15.2 point 3). An event dispatched again with seq <= redeliverUpTo carries redelivered: true; an event never dispatched before in this window does not. A new identify or resume starts a fresh window without the flag. Ruling for docs/DECISIONS.md: the flag marks exactly the ack-timeout redeliveries, as EventEnvelope.redelivered describes; the deadline runs from the oldest unacknowledged dispatch.
    • Frames: while the window is in the backlog phase it sends cursor {seq: read} after every page whose read is above the last seq it sent, so an adapter can acknowledge past events it may not see (R-SY-03); when read first reaches head_seq of ready it sends catchup.complete {head_seq} once and enters the live phase; live it sends cursor whenever the skipped numbers reach sync.cursor_advance_every (04 7.1, 7.2 point 4, as M1-65 does for people). Ruling for docs/DECISIONS.md: a consumer connection always catches up from acked_seq + 1 and ends that catch-up with catchup.complete.
    • Resync: at start the runner calls CatchUp.Check with after = acked. retention (events after acked were removed) and cursor_ahead (only after a restore without reset) send resync_required {reason, head_seq, sync_epoch}, set the cursor to that head_seq with Restart, and continue live from there without catchup.complete, as M1-65 does after resync_required. too_far_behind does not apply to consumers: the backlog is delivered through the window, because agent events are commands and the window already bounds the load; epoch_changed cannot occur, because the client epoch is ignored and the restore resets agent cursors (04 7.5 point 4, M7-04). Ruling for docs/DECISIONS.md.
    • Purge: when the queue purge of M1-65 drops dispatch frames of a removed target, the window drops their numbers from inflight so they hold no budget; the next cursor frame lets the adapter acknowledge past them.
  6. Runner (gateway/app). One goroutine per consumer connection. The hub does not queue durable dispatch frames for agent sessions; after each event it wakes their runners with a non-blocking, coalescing signal. The runner reads pages with the sync port in authorize.Delivery(ctx) mode (M1-65), page size sync.catchup_page_size, upper bound the hub position: the highest seq the process's log tail has passed through Authorizer.Observe and Hub.Dispatch. Reason: every admission then happens with a permission cache that has seen every access.changed up to that event (04 8.1, R-SY-02), and a full window makes the runner wait instead of filling a buffer that would close the connection with 4014. Envelopes carry DeliveryFields{IncludeTraceContext: true, Redelivered: ...} (M1-16, M1-63). The runner waits for queue space like the catch-up runner of M1-65 and retries a failing page read with backoff capped at database.listen_reconnect_backoff_cap while the connection is open (02 14.4). The access.changed handling of M1-65 (revalidation with 4012, index update, purge) applies to agent sessions unchanged. Ruling for docs/DECISIONS.md: consumer connections pull from the log up to the hub position.
  7. DeliveryHook (outgoing port of gateway/app): Prepare(ctx context.Context, agentID ids.ID, ev sync.Envelope) (sync.Envelope, bool, error). Every dispatch to a consumer, on both transports, passes it; false means "do not deliver, treat as checked". This ticket binds passdelivery; M2-08 binds runs.Delivery there (02 10).
  8. Acknowledgement rules (both transports; agents/domain.CheckAck): seq below 0 or above the head of note 9 is validation_failed (pointer /seq); seq at or below the stored or held acknowledgement changes nothing and is no error (06 2 "No effect"; monotone, 06 7); otherwise the acknowledgement advances. Acknowledging beyond what was dispatched is allowed and skips those events, because an acknowledgement covers every number up to seq (04 15.2 point 1). Gateway op ack {seq}: the dispatcher handler calls agents.Consumers.AcceptAck (authorize agent.events.consume, then CheckAck), then applies the acknowledgement to the session's window and marks it dirty; nothing is written yet; no reply frame (asyncapi receive.ack has none). Errors answer error {op: "ack", code} and keep the connection (04 6.2): validation_failed, forbidden (people and apps, subject_kind_not_allowed), rate_limited for class rl.agent_runtime (06 9 point 3; if the merged dispatcher has no per-op rate limit yet, apply the class to ack with platform/ratelimit, keyed by the agent). Ruling for docs/DECISIONS.md: the upper bound and the "no change" rule.
  9. Head and saving. The head that consumer operations report and check acknowledgements against is the hub position of the serving process (note 6); only a gateway connection creates a missing cursor at the head_seq of its own ready (note 4 b), because that is the head the client was told. Ruling for docs/DECISIONS.md: a consumer never acknowledges or is told about a seq this process cannot yet deliver. Saving: a process-wide ticker with interval agent.ack_persist_interval calls Save for every consumer session whose held acknowledgement is above its last saved value. ErrConsumerReplaced closes that connection with 4015 and ends the session (the takeover happened elsewhere, 04 15.1 point 3); other errors keep the session dirty for the next tick (log WARN). Flush with Save also before a takeover in this process (note 4 a), when a connection detaches or closes for any reason (including 4012 and 4014), and in step 3 of the shutdown before CloseAll(4013) within the remaining shutdown.total_timeout budget (02 15.3). Acknowledgements made after the last successful save of a connection that loses the row are lost; those events come again (at least once, 04 9; the adapter deduplicates by event ID, 04 15.2 point 4).
  10. REST operations (agents/app, each first authorize(actor, agent.events.consume, {kind: agent, id: actor, owner: actor}), class rl.agent_runtime):
    • getConsumerState: the stored row; 404 not_found if this consumer never connected (reading does not create a cursor; 04 15.1 point 2 creates it at the first connection). Answer ConsumerState {consumer, acked_seq, head_seq} with head_seq from ConsumerFeed.Head.
    • pollAgentEvents: delegates to ConsumerFeed.Poll. The engine keeps one REST window per ConsumerKey in the process with its own holder id (a UUIDv7 created with the window). Per call, under the per-key lock: flush and close a gateway session of the key in this process with 4015 (note 4 a and c); Take(consumer, holder = REST holder, head) (the first poll creates the cursor at head_seq and returns an empty page); reconcile the window with the stored acked_seq (drop inflight entries up to it); apply the timeout rule of note 5; resync check as note 5, where retention and cursor_ahead move the cursor to head_seq with Restart and answer 410 resync_required with reason; otherwise read up to the budget, where limit defaults to agent.max_inflight_events and larger values are capped to it (06 6 point 3). Answer EventPage {events, next_cursor, head_seq, sync_epoch} with next_cursor = the window's read (the adapter acknowledges it after processing, also when events is empty). Within agent.ack_timeout a poll never repeats inflight events; it only continues after them while the budget allows. A REST window idle for sync.resume_window is dropped by the sweeper; the next poll starts a fresh window from acked_seq + 1. Ruling for docs/DECISIONS.md: a REST poll is a consumer like an identify (the newest wins, R-SY-09), the per-process window, the idle rule, and 410 for retention and cursor_ahead.
    • ackAgentEvents: head from ConsumerFeed.Head, CheckAck, then AckConsumer (acked_seq = GREATEST(acked_seq, $seq) immediately, without holder condition, 404 not_found for an unknown consumer), then ConsumerFeed.Acked so a window of the key in this process sees it; answer ConsumerState. Neither getConsumerState nor ackAgentEvents takes the consumer over. Ruling for docs/DECISIONS.md.
  11. Permissions. The descriptor of note 10 makes AgentSelf decide (M1-28): an agent passes for itself; a person or app token gets 403 forbidden with reason: subject_kind_not_allowed and no row is created. Take and AcceptAck use the same descriptor. Denials are audited by authz (R-AZ-04); admission of events in the runner and the REST poll runs in delivery mode and writes no audit entry (07 4.2, AZ-P13). No permission version, no access.changed, no event of any kind is written by this ticket (05: Permissions no).
  12. Errors (06 5; codes only from x-problem-codes):
Situation REST Gateway
consumer does not match the pattern 400 validation_failed, pointer #path/consumer identify closes with 4002
caller is not an agent 403 forbidden, reason: subject_kind_not_allowed ack answers error with forbidden; consumer ignored at identify
consumer never connected getConsumerState, ackAgentEvents: 404 not_found not applicable (identify creates)
seq below 0 or above the head 400 validation_failed, pointer /seq error with validation_failed
seq at or below the acknowledgement 200 with the current state nothing
limit below 1 400 validation_failed, pointer #query/limit not applicable
retention, cursor_ahead 410 resync_required with reason; cursor moved to head_seq resync_required {reason, head_seq, sync_epoch}; cursor moved
class rl.agent_runtime exhausted 429 rate_limited with Retry-After error with rate_limited
another consumer took over not applicable older connection closes with 4015
  1. Concurrency. One runner goroutine per consumer connection owns the window; the reader goroutine passes acknowledgements to it (channel or the window's mutex), and the save ticker reads the held value under the same lock. A per-process lock per ConsumerKey serializes takeovers by identify, resume and REST polls. Across processes the last Take wins and the loser learns it at its next save (fencing); an acknowledgement through ackAgentEvents in another process is monotone and cannot lower acked_seq. Two consumer names of the same agent never affect each other.
  2. Configuration keys used: agent.ack_timeout, agent.ack_persist_interval, agent.max_inflight_events (registered by this ticket), sync.catchup_page_size, sync.cursor_advance_every, sync.resume_window, sync.outbound_queue_max, gateway.write_timeout, database.listen_reconnect_backoff_cap, shutdown.total_timeout, ratelimit.rl.agent_runtime.per_minute, ratelimit.rl.agent_runtime.burst.
  3. Rulings for docs/DECISIONS.md (one row each with reason and place): note 1 (one engine, ConsumerFeed); note 2 (fencing always on); note 3 (pattern, agents only); note 4 (takeover order, session end after 4015, client cursor ignored, 4015 on resume); note 5 (flag and deadline, catch-up with catchup.complete, resync reasons for consumers); note 6 (pull up to the hub position); note 8 (upper bound, no change at or below); note 9 (head of the serving process); note 10 (REST takeover, window, idle rule, 410, no takeover by state and ack). The first connection at head_seq, the inflight limit, the redelivery from acked_seq + 1 and 4015 are in 04 and need no ruling.

Tests (write them first)

Unit (go test, fake clock from kernel/clock, fakes for ports):

  • internal/gateway/domain: TestConsumerWindow_BudgetBoundsInflight (no read at zero budget; a page with more admitted events than the budget is cut and read is the last dispatched seq); TestConsumerWindow_AckCumulativeAndMonotone (acknowledging n drops every inflight entry up to n; a lower value changes nothing); TestConsumerWindow_TimeoutRestartsFromAcked (after the deadline reading restarts at acked + 1, exactly the previously dispatched numbers carry the flag, numbers never dispatched do not, the deadline follows the oldest inflight entry after a partial ack); TestConsumerWindow_PurgeReleasesBudget; TestConsumerWindow_CursorAndCatchupFrames (backlog: cursor after each page with skipped numbers; catchup.complete exactly once when read reaches the ready head; live: cursor after sync.cursor_advance_every skipped numbers); TestConsumerWindow_ResyncReasons (retention and cursor_ahead restart at the head, too_far_behind is ignored).
  • internal/agents/domain: TestValidConsumerName (default, allowed and rejected values); TestCheckAck (below 0, above head, equal, below, above).
  • internal/agents/app with fake repository and fake authorizer: TestTakeCreatesAtHead, TestTakeKeepsAckedAndWritesHolder, TestSaveFencedReturnsReplaced, TestRestPollDelegatesAfterAuthorize, TestAckEventsUnknownConsumerNotFound.
  • internal/gateway/app with fakes for agents.Consumers, the sync port, the hub position and a writer that blocks: TestConsumerRunner_WaitsAtMaxInflight, TestConsumerRunner_ReadsOnlyUpToHubPosition, TestConsumerRunner_RedeliversAfterAckTimeout, TestConsumerRunner_DeliveryHookCanDrop (an envelope the hook rejects is not sent and counts as checked), TestConsumerTakeover_FlushTakeCloseOrder (save of the old session, then Take, then 4015, then the old session is gone), TestConsumerSave_ReplacedCloses4015, TestConsumerFlush_OnCloseAndShutdown, TestRestWindow_NoRepeatWithinTimeout, TestRestWindow_IdleSweptAfterResumeWindow, TestAckOp_ErrorKeepsConnection.

Integration (PostgreSQL 18 via pgtest, the in-process server and the wstest client of M1-64 as in M1-65's tests, HTTP through the real chain; Owner Alex with step-up creates the agent support-agent through the agents port of M1-51 and issues its API token with a DPoP key from identitytest (M1-44); Alex grants the agent read on channel Allgemein of space Support with authz.Admin.CreateGrant (M1-29); Sam posts messages with createMessage (M1-56) in Allgemein (visible to the agent) and in a channel of space Marketing without a grant (invisible); updateAgent by Alex produces agent.updated about the agent):

  • TestConsumerFirstConnectionStartsAtHead (plan acceptance "the first connection starts at head_seq"): messages posted before the first identify are never delivered; the row exists with acked_seq equal to ready.head_seq; the first dispatch after catchup.complete {head_seq} has a seq above it; the same holds for the first pollAgentEvents of another consumer name (empty page, next_cursor and acked_seq equal head_seq).
  • TestConsumerContinuesFromAckedSeq: acknowledge some events, close, post more, identify again: delivery starts at acked_seq + 1; acknowledged events do not come again; unacknowledged ones come without redelivered.
  • TestConsumerMaxInflightEvents (plan acceptance agent.max_inflight_events, key set low in the test configuration): post more visible messages than the key allows; exactly that many dispatch frames arrive and no more while nothing is acknowledged; acknowledging k of them releases exactly k more; pollAgentEvents with a larger limit returns at most that many events.
  • TestConsumerRedeliveryAfterAckTimeout (plan acceptance "without ack, delivered again after agent.ack_timeout with redelivered", short timeout): without acknowledgement the same events come again from acked_seq + 1 with redelivered: true, the first delivery has no flag, events dispatched for the first time after the reset have no flag; after acknowledging nothing comes again; over REST a second poll inside the timeout returns no repeated event and a poll after it returns them with the flag. Every frame decodes into its generated type with unknown members rejected, and the stored event_log rows contain no redelivered member.
  • TestConsumerSecondIdentifyCloses4015 (plan acceptance "a second identify of the same consumer closes the old one with 4015"): two connections with the same token and consumer; the first closes with 4015 and reason consumer_replaced; acknowledgements it made before are stored; later events reach the second; holder_session_id equals the second's gateway_session_id; a resume of the first session gets 4006; a connection of the same agent with another consumer name stays open and keeps its own cursor.
  • TestConsumerFencedSaveAcrossInstances: two server instances on one database (as AZ-P9 in M1-65); the agent connects to instance A, then to instance B with the same consumer; after the agent acknowledges on A, A's connection closes with 4015 within agent.ack_persist_interval plus test slack, and acked_seq is not changed by A's late save.
  • TestConsumerAckPersistedWithinInterval and TestConsumerShutdownFlushesAck: an acknowledgement is visible through getConsumerState within agent.ack_persist_interval; an acknowledgement right before shutdown is stored and the connection closes with 4013.
  • TestConsumerStreamIsFilteredByRights (07 8.1, AZ-P9 for consumers): the agent receives message.created of Allgemein and agent.updated about itself, never messages of Marketing; cursor frames cover the skipped numbers; after Alex revokes the grant on Allgemein (authz.Admin.RevokeGrant) through server instance A while the agent is connected to instance B (two instances on one database, as AZ-P9 in M1-65), the agent receives access.changed, no later message of Allgemein, and the redelivery after agent.ack_timeout does not repeat an unacknowledged Allgemein message.
  • TestConsumerAckValidation: ack above the head answers error with validation_failed and the connection stays; ack below the acknowledgement changes nothing; ackAgentEvents above the head gives 400 validation_failed with pointer /seq, a negative seq gives 400, an unknown consumer 404, a valid one 200 ConsumerState with the stored value.
  • TestPollTakesOverGatewayConsumer: with a gateway connection holding consumer default, pollAgentEvents for default closes it with 4015 and delivers; getConsumerState and ackAgentEvents close nothing.
  • TestPollRetentionGivesGone: after partition maintenance of M1-16 removed the events after a consumer's acked_seq (short sync.event_retention in the test), pollAgentEvents answers 410 resync_required with reason: retention, acked_seq equals the reported head, and the next poll answers 200; the same consumer over the gateway gets resync_required with retention and stays open.
  • TestConsumerNameRules: identify with an invalid consumer closes with 4002; a REST path with an invalid name gives 400 validation_failed; identify without consumer uses default.
  • TestPeopleAndAppsUnchanged: Sam's connection still receives live events as in M1-65 and has no cursor row; an app identify with consumer creates no row; Sam's ack answers error with forbidden; Sam's token on the three REST operations gives 403 forbidden with reason: subject_kind_not_allowed.
  • TestConsumerRoutesAndOpRegistered (AZ-P1, AZ-P2): the three routes are registered with agent.events.consume and rl.agent_runtime; the op ack is registered by module agents and is in gateway_ops.enabled; the start check of M1-25 passes.
  • TestConsumerDenialsAudited (AZ-P13): a denied REST call writes one aggregated deny entry; the runner's delivery filtering writes none.
  • TestConsumerAgentDisabledCloses4012: disableAgent closes the consumer connection with 4012 after its acknowledgement was saved.

Commands that must pass, from the repository root:

  • cd services/server && go test -race -count=1 ./internal/agents/... ./internal/gateway/... ./internal/sync/... ./internal/authz/... ./internal/kernel/... ./internal/wire/... ./migrations/...
  • make migrations-lint, make codegen-check, make arch, make register-check, make api-lint, make api-checks, make config-doc-check, make docs
  • make check

Acceptance criteria

  • agent_consumer_cursors exists exactly as 03 defines it; only agents reads and writes it; make migrations-lint and make register-check are green.
  • getConsumerState, pollAgentEvents, ackAgentEvents and the op ack are enabled and registered with agent.events.consume; AZ-P1 and AZ-P2 pass.
  • The first connection of a consumer, over the gateway and over REST, begins at head_seq; later ones at acked_seq + 1.
  • At most agent.max_inflight_events unacknowledged events are on the way per consumer; acknowledging releases exactly as many.
  • Without ack within agent.ack_timeout the events come again from acked_seq + 1 with redelivered: true; the flag never reaches the event log.
  • A second identify of the same consumer closes the older connection with 4015; a save that finds another holder closes with 4015, also across two server instances; other consumers of the agent are unaffected.
  • acked_seq is saved within agent.ack_persist_interval, on close and at shutdown before 4013.
  • Every delivery and redelivery is admitted with the permissions at delivery time; delivery filtering writes no audit entry, denials of requests and ops do.
  • docs/api/openapi.yaml, docs/spec/09-agents.md, docs/spec/02-architecture.md, the READMEs and docs/DECISIONS.md describe what was built.
  • make check is green.

Definition of done

  • Code and tests as specified; doc comment on every public symbol (purpose, parameters, return value, errors, example where useful); comments explain why.
  • READMEs of agents and gateway updated (status, ports Consumers and ConsumerFeed, table, action, op ack, configuration keys, the window, takeover, save ticker and shutdown flush).
  • Rulings of Design note 15 recorded in docs/DECISIONS.md (one row each with reason).
  • services/server/CHANGELOG.md entry under 'Unreleased'.
  • If the implementation shows a threat that docs/security/threat-model.md 5.10 does not list (for example the takeover of a consumer with a stolen agent token), add the row with its countermeasures (DPoP binding, revocation closes with 4012) and register T-TK-02.
  • An ADR only if you take an architecture decision beyond this brief (MADR 4 as in docs/adr/0000-template-madr-4.md, headings and text in English, next free number, make adr-index).
  • make check green.
  • An independent code review over the diff, then the red team gate (docs/quality/red-team-gate.md, verdict JA or NEIN); on NEIN fix with a test per finding and have the delta checked again. Record both verdicts.
  • Land the change as AGENTS.md ("How work lands on main") describes: a pull request from this branch, squash-merged only when review, red team and make check, ci-ok and the review threads are green; then git worktree remove.

Report

Write to {{REPORT_PATH}}:

  1. First line exactly Verdict: DONE or Verdict: BLOCKED <reason>.
  2. Commits (SHA and subject), changed files with line counts, test results with counts (the red run before the change and the green run after), the summary of make check.
  3. Review and red team verdicts and the findings fixed.
  4. What is not proven (for example timing margins of the timeout and save tests, behaviour with cluster.mode = multi beyond the two-instance test).
  5. Follow-ups that could not be done in this scope, each with its reason (at least the metric bothy.agent.consumer.lag without owner), and spec gaps found.

Then write {{DONE_PATH}} with exit=0 sha=<sha> or exit=1 <reason>.