diff --git a/PR-4-REQUIREMENTS.md b/PR-4-REQUIREMENTS.md new file mode 100644 index 0000000..c7ec38c --- /dev/null +++ b/PR-4-REQUIREMENTS.md @@ -0,0 +1,239 @@ +# PR 4 — Human-paced chunked delivery + +Today a turn returns its complete reply and the channel sends it as one message +(`internal/core/core.go`, `handler.Received`; `internal/agent/process.go`, +`process.prompt` collects every `agent_message_chunk`). This PR lets a channel +deliver the reply while the turn runs, split into paragraphs, one message at a +time, with a pause before each paragraph that is proportional to its length. +Splitting and pacing are off by default, and each channel is configured on its +own. + +Branch: `feat/paced-delivery`. One PR, roughly 250-350 lines with tests. +Commit line: `feat(core): deliver a reply in paced paragraphs`. + +## Behaviour + +1. The agent reports each chunk as it arrives, instead of one complete reply. +2. Text chunks accumulate in a buffer. The channel's separator closes a + paragraph. An empty separator, which is the default, closes nothing, so the + buffer holds the whole reply. +3. A closed paragraph is queued and delivery starts at once, while the agent is + still generating: the first paragraph reaches the contact as soon as its + separator arrives, whatever the rest of the reply takes. +4. The queue delivers in order. Before each paragraph it waits + `words × 60s / words_per_minute`, capped at `max_delay`, minus the time + already spent since the previous paragraph was delivered, and never less than + zero. The agent's own generation time is therefore part of the pace: a + five-word paragraph at 60 words per minute has a five-second target, and if + the agent took four seconds to produce it the queue waits one second. + The first paragraph of a turn measures from the start of the turn. +5. A non-text block (image, audio, resource, resource link) closes the pending + text buffer first, then is delivered as its own message with no pause. +6. When the turn ends, the pending buffer is closed, queued, and delivered under + the same pace. +7. Leading and trailing whitespace of a paragraph is removed; an empty paragraph + is dropped and makes no call to `Reply`. +8. `Received` returns only after the last paragraph is delivered, so the channel + still reports a failed turn on its own transport. +9. A failed `Reply` ends the turn's delivery: the remaining paragraphs are + dropped and `Received` returns the error. +10. A cancelled context ends the delivery, drops what is pending, and returns + `ctx.Err()`. Nothing is sent after cancellation. + +With the defaults, rules 2 and 4 collapse into the behaviour momo has today: one +message per turn, sent as soon as the turn ends. + +## Contract changes + +`internal/core/core.go`: + +```go +// Emit delivers one part of a turn's reply. The agent calls it as the content +// arrives. It does not block and returns the error of an earlier part, so an +// agent talking to a broken channel stops generating. +type Emit func(content []ContentBlock) error + +type Agent interface { + Turn(ctx context.Context, m Message, emit Emit) error +} +``` + +`Agent.Turn` no longer returns content. Remove the old signature; keep no +compatibility path. + +`internal/agent`: `process.prompt` calls `emit` for each `agent_message_chunk` +update it accepts while collecting, and collects nothing. The `collected` field +and its use go away. The chunk arrives on the JSON-RPC handler goroutine, so +`emit` must not be called under `process.mu`. + +`internal/core/delivery.go` (new): the paragraph splitter, the pace, and the +ordered delivery queue. It is the only place that knows about pacing. + +## Configuration + +Delivery is a property of the channel, not of momo: a contact on WhatsApp reads +paced paragraphs, an ACP client wants the reply as fast as momo can send it. Each +channel block takes its own `delivery` block, and the defaults keep both +splitting and pacing off: + +```yaml +channels: + respondio: + api_token: "paste the respond.io API access token" + delivery: + # Text that closes a paragraph. Empty means the reply is one message. + # Default: "" + separator: "\n\n" + # Pace of the pause before each paragraph. 0 means no pause. + # Default: 0 + words_per_minute: 60 + # Cap on one paragraph's pause. Must be positive. Default: "10m" + max_delay: "10m" +``` + +Refused at start-up, with a message naming the channel and the setting: a +negative `words_per_minute`, and a non-positive `max_delay`. A `separator` set +while `words_per_minute` is 0 is valid: the reply is split and sent with no +pause. + +`delivery` is a key of a channel's block that the channel implementation never +sees. `internal/config/config.go` takes it out of the block before it builds the +channel's `Decoder`, so a channel's own strict decoding is unaffected, and hands +it to `internal/channel`. `channel.Build` decodes it through the core, and wraps +the `core.Reply` it passes to `Received` in the delivery the settings describe — +next to `qualifying`, which already wraps the handler for every channel. No +channel implementation changes in this PR, and a channel added later gets +delivery without doing anything. + +## Structure + +- Delivery holds one goroutine for the turn, fed by a queue of ready parts: + a mutex, a slice, and a signal channel of capacity one. Ordering comes from the + single consumer, not from a lock around `Reply`. +- `emit` never blocks, because `jsonrpc2.Conn` reads and dispatches serially + (`conn.go`, `readMessages`): a blocked handler stops the connection and with it + the `session/request_permission` of the same turn. A backlog grows in the queue + instead, delivery continues after `Turn` returns, and that is why `Received` + waits for the queue to drain. +- The pause is a pure function of the paragraph, the two settings, and the time + already spent. The queue goroutine reads the clock once per paragraph, at the + moment it takes it from the queue, and waits with `select` on `ctx.Done()` and + `time.After`. No injected clock: the pause function is tested with literals, + and the ordering tests run with `words_per_minute: 0`. +- Splitting text into paragraphs is a pure function of the buffer: no IO, no + clock, no goroutine. Test it on its own. + +## Tests (write first) + +`internal/core/delivery_test.go`, with a fake `Reply` recording calls: + +- `pause`, called directly, with literal expectations and no configured value in + the assertion: five words at 60 words per minute and nothing spent gives `5s`; + the same paragraph with `4s` spent gives `1s`; with `6s` spent gives `0`; a + 200-word paragraph with `max_delay: 10s` gives `10s`; any paragraph with + `words_per_minute: 0` gives `0`. +- The queue honours the pause: one word at 1200 words per minute makes + `Received` take at least `50ms`. + +The rest with `words_per_minute: 0`: + +- Two paragraphs in one chunk produce two `Reply` calls, in order. +- The first paragraph is delivered before the second chunk is emitted: the fake + `Reply` records the first call while the test still holds the generator. +- A generator that emits ten paragraphs at once still delivers them in order, + and `Received` returns only after the tenth `Reply` call. +- Three chunks with no separator between them produce no `Reply` call until the + turn ends; the end flushes one `Reply` call whose text is the three chunks + concatenated, with nothing inserted between them. +- The separator inside one chunk and across a chunk boundary both close a + paragraph. +- A configured `separator: "---"` splits on `---` and leaves `\n\n` in the text. +- The default settings deliver a reply holding two blank lines as one `Reply` + call, at the end of the turn. +- Whitespace-only content makes no `Reply` call. +- An image block flushes the pending text first, then goes out with no pause. +- A `Reply` that fails stops delivery: no further `Reply` call, the next `emit` + returns that error, and it reaches the caller of `Received`. +- A cancelled context stops delivery and returns `ctx.Err()`. + +`internal/channel/channel_test.go`: a channel configured with a `delivery` block +does not see that key in its own settings decoding; two channels configured with +different separators split their replies differently in the same run. + +`internal/config/config_test.go`: the defaults when the block is absent, and the +two refusals above. + +`internal/agent/harness_test.go`: the stub agent +(`internal/agent/testdata/stubagent`) sends two chunks; the test asserts two +`emit` calls before `Turn` returns. + +End to end (`cmd/momo/main_test.go` style): a webhook whose stub reply holds two +paragraphs results in two channel messages when the channel configures a +separator, and one message when it does not. + +## The ACP channel side + +The specification allows it ([Prompt Turn](https://agentclientprotocol.com/protocol/v1/prompt-turn), +sections "Agent Reports Output" and "Check for Completion"). A prompt turn is a +stream: the agent reports the model's output with `session/update` notifications +and only then answers the `session/prompt` request with a `StopReason`. Many +notifications per prompt are the normal case. momo's ACP channel already sends +one `agent_message_chunk` notification per block, and `handler.Received` drains +the delivery queue before the endpoint answers with `end_turn`, so every update +is sent before the response, as v1 requires. + +One addition belongs to this PR, because this PR is what lets a turn produce +more than one message: v1 carries an optional `messageId` on a message chunk +([Message IDs](https://agentclientprotocol.com/protocol/v1/prompt-turn#message-ids)). +Chunks with the same `messageId` are one message; a changed `messageId` starts a +new one. Without it a client joins every paragraph into one message and the +splitting is invisible. + +- `internal/acp/acp.go`: `Update` gains a `MessageID` field, JSON name + `messageId`, omitted when empty. +- `internal/channel/acp/methods.go`: `reply` assigns one fresh id per call, so + each paragraph is its own message and every block of one call shares that id. +- Test: two `Reply` calls produce two different `messageId` values; the blocks of + one call share one value. + +## README + +`README.md` describes what momo does today, so it changes with this PR. Keep the +tone of the file: plain statements of behaviour, one idea per sentence, every +setting commented with its default in the configuration sample. + +- Configure: add the `delivery` block inside a channel block of the sample, with + a comment per setting stating what it does and its default, in the form the + other settings use. State that every channel takes the block and that the + defaults send the reply as one message. +- Connect an agent: rewrite steps 4 to 6 of "One message is one turn". The agent + streams its answer, and momo delivers it as the channel's delivery settings + say: one message at the end of the turn by default, or one message per + paragraph while the turn runs. +- Connect an agent: add a short paragraph after the turn steps that states the + splitting rule, the pause and how the agent's generation time counts towards + it, that the pause never blocks the agent, and that a send that fails stops the + turn. +- Connect an ACP client: state that a turn produces one `session/update` + notification per delivered message, each with its own `messageId`, and that the + `session/prompt` response with the stop reason follows the last of them. + +## Out of scope + +- The per-conversation FIFO inbox, steering and cancellation of a stale turn + (PR 5). The `Harness.acquire` mutex stays as it is. +- Splitting on a pattern rather than a literal separator (regular expressions, + sentence detection, length limits). +- Retry of a failed `Reply`. + +## References + +Agent Client Protocol, version 1 (checked 2026-02): + +- [Prompt Turn](https://agentclientprotocol.com/protocol/v1/prompt-turn) — the + update stream during a turn, `messageId` on message chunks, stop reasons, and + the rule that pending updates are sent before the `session/prompt` response. +- [Content](https://agentclientprotocol.com/protocol/v1/content) — the block + types `agent_message_chunk` carries. +- [Schema](https://agentclientprotocol.com/protocol/v1/schema) — field names and + required fields of `session/update`. diff --git a/PR-5A-REQUIREMENTS.md b/PR-5A-REQUIREMENTS.md new file mode 100644 index 0000000..3ed9884 --- /dev/null +++ b/PR-5A-REQUIREMENTS.md @@ -0,0 +1,257 @@ +# PR 5a — One turn at a time for one conversation + +Two rules move into place with this PR. + +A conversation runs one turn at a time, including the delivery of that turn's +reply. Messages that arrive while it is busy are collected and reach the agent +together, as the prompt of the next turn. Today +`internal/agent/harness.go` holds a map of one channel per conversation, and +`Harness.acquire` makes a second message wait for its own turn behind the first. + +Everything an agent generated is delivered. Today `internal/core/delivery.go` +drops the paragraphs still queued when the turn's context is cancelled, and the +harness never learns of it: the session history holds a reply the contact never +received. + +Branch: `feat/serial-turns`. +Commit line: `feat(core): run one turn at a time for one conversation`. + +## Why the core and not the harness + +"One turn at a time for one conversation" is a statement about conversations. It +is true whatever runs the turn, so a second `core.Agent` implementation would have +to write it again. It becomes a `core.Handler` decorator, next to `Delivery`, and +it is then testable against a fake agent with no subprocess, no pipes and no ACP. + +`agent.Harness` loses `acquire`, the `turns` map and the mutex that guards it, and +with them the leak the `ponytail:` note on `acquire` describes. The harness keeps +`turn_timeout`, which still bounds one turn. + +The decorator wraps the delivery rather than sitting under it. A conversation is +busy from the arrival of a message until the last paragraph of the reply has been +sent, so two replies of one conversation can never be paced out at the same time, +and a message that arrives during the pacing joins the next turn. When a turn +starts, the contact has received everything the agent's session history holds. + +## Behaviour + +### Turns of one conversation + +1. A message for a conversation that is not busy starts a turn at once, carrying + that message's content only. +2. A message for a busy conversation is appended to that conversation's pending + batch, and its caller waits. A conversation is busy from the moment a turn + starts until the reply of that turn has been delivered in full. +3. When the conversation becomes free and the batch is not empty, one further turn + runs with the whole batch as its prompt: the content blocks of every message in + the batch, concatenated in the order the messages entered it, with nothing + inserted between them. +4. Every caller whose message was in that batch returns when that turn ends, and + returns that turn's error. The caller that started the turn in rule 1 returns + its own turn's error, not the batch's. +5. Batches repeat: messages arriving during the batch's turn form the next batch, + and the conversation keeps running turns until no message is pending. +6. Turns of different conversations never wait for each other. +7. A caller whose context is cancelled while it waits returns `ctx.Err()`. Its + content stays in the batch and is still sent to the agent: a message momo + accepted is not dropped because the transport that brought it gave up. +8. A turn that fails leaves the conversation free like a turn that succeeded, so a + pending batch is never stranded behind an error. +9. The conversation's state is removed when it is free and nothing is pending, so + the map holds only the conversations that are busy. + +The turns of one conversation run on the goroutine of the caller that found the +conversation free. That caller stays inside `Received` until the conversation is +quiet, and every turn it runs uses its context and its `Reply`. The batch is not +handed to a waiting caller, because a waiting caller may have given up, while the +caller that is running is blocked in `Received` and is therefore able to deliver. +Consequences to state plainly: + +- The reply of a merged turn is delivered on the transport of the caller that ran + it. For respond.io that is the same contact either way. For an ACP client that + prompted twice on one session over two connections, the reply reaches the + connection of the first prompt. +- A waiting caller's own `Reply` is never called, so a channel delivers a merged + reply once, not once per merged message. +- A message waits at most two turns and their deliveries: the turn running when it + arrived, and the batch's own. No queue depth and no new setting is needed. + +### Nothing generated is dropped + +10. A part that is queued for delivery is delivered, whatever happens to the + context of the turn that produced it. The paced pause is kept. +11. Delivery ends before the queue is empty in one case only: a `Reply` that + fails. The parts still queued are dropped and the turn returns that error, + because the transport that would carry them is the one that just failed. +12. The queue sends with a context that cannot be cancelled, so a turn whose + deadline passed still delivers what it generated instead of failing every + remaining send. + +Rule 12 makes a stopped momo wait for the drip it has queued, bounded by +`max_delay` per paragraph, and `shutdown_timeout` cuts it. Draining at shutdown is +PR 5d's subject. + +## Contract changes + +None in the interfaces. `core.Agent`, `core.Handler`, `core.Emit` and +`core.Reply` keep their signatures. + +`internal/core` gains one exported function: + +```go +// Serialize runs one turn of a conversation at a time, delivery included, and +// sends the messages that arrive meanwhile as the prompt of the next turn. +func Serialize(log *slog.Logger, h Handler) Handler +``` + +`internal/channel/channel.go`: `Build` wraps every channel with it, outside the +delivery, so no channel and no assembly can forget it: + +```go +h := core.Serialize(log, delivery.Handler(qualifying{name: name, h: h})) +``` + +`Build` therefore takes a `*slog.Logger`. `cmd/momo/main.go` passes the one it +already holds. + +One `Serialize` per channel instance, keyed by that channel's own conversation +id. A conversation is one channel and one contact, so two channels never share a +key and no map is shared between them. `qualifying` stays inside, which is why the +key is the unqualified id. + +`internal/core/delivery.go`: the queue no longer selects on the turn's +`ctx.Done()` for a part it already holds, and it calls `Reply` with +`context.WithoutCancel` of the turn's context. The `fail` path stays for a failed +`Reply`. + +One turn at a time is what momo promises, not a mode an operator selects, so no +setting is added. + +## Structure + +`internal/core/conversation.go`, one file, holding: + +- A map from conversation to its state, guarded by one mutex. Every arrival, every + release and the removal in rule 9 take that mutex, so two arrivals for one + conversation can never end up with two states. +- Per conversation: whether it is busy, and the pending batch or nothing. +- Per batch: the content collected so far, a channel that is closed when the + batch's turn has ended, and that turn's error, written before the channel is + closed. +- The caller that finds the conversation free runs a loop: call the inner handler, + take the pending batch under the lock, call the inner handler again, until there + is no batch, then mark the conversation free and remove its state. +- A waiting caller selects on the batch's channel and on its own `ctx.Done()`. +- One log line when a message joins a batch, so no accepted message is invisible + in the log. + +No goroutine is started, no channel is used as a queue, and no `atomic` type is +needed. The inner handler is called outside the lock; nothing else is. + +## Tests (write first) + +`internal/core/conversation_test.go`, with a fake `Handler` whose `Received` +blocks until the test releases it and records the message it was given: + +- A second message does not start a second turn while the first runs: the fake + sees one call while the first is held. +- Two messages arriving during one turn produce exactly two calls in total, and + the second call's content holds the blocks of both messages in arrival order. +- Turns of two conversations run at the same time: hold both, release neither, and + both are inside the fake before either returns. The test fails by deadlock and + not by a measurement of time. +- A waiting caller returns only after the merged turn ended, and returns that + turn's error. +- The caller that started the first turn returns the first turn's error, and not + the merged turn's. +- The merged reply is delivered through the `Reply` of the caller that ran the + turn: each caller passes a `Reply` of its own, and only that one records + content. +- A turn that fails still runs the batch behind it. +- A waiting caller with a cancelled context returns `context.Canceled`, and its + content is still in the next call. +- After every turn has ended, the map holds no entry for the conversation. The + test is in package `core`, so it reads the map under the same mutex. +- A message that arrives after the conversation went quiet starts a turn of its + own, with its own content only. + +`internal/core/delivery_test.go`: + +- A turn whose context is cancelled after two of its four paragraphs are queued + still delivers all four, in order. +- The context the queue passes to `Reply` is not cancelled, although the turn's + context is. +- A `Reply` that fails still drops the parts behind it and returns its error, as + it does today. +- The existing test that a cancelled context stops delivery and returns + `ctx.Err()` states the behaviour this PR removes. Replace it with the two above; + do not keep both. + +`internal/core/conversation_test.go`, on the two rules together: a channel that +paces its delivery, a turn holding two paragraphs, and a message arriving after +the first paragraph was sent. The message reaches the fake agent only after the +second paragraph was delivered, and the two paragraphs are not interleaved with +the reply of the second turn. + +`internal/agent/harness_test.go`: keep +`TestTurnsOfDifferentConversationsRunAtTheSameTime`. It now states a property of +the harness — two subprocesses, two sessions, no shared state — and it fails if +the harness ever takes a lock again. + +`internal/channel/channel_test.go`: two messages of one conversation, delivered +through a built channel, produce one merged turn; two messages of two +conversations do not wait for each other. + +`cmd/momo/main_test.go`: two webhooks for one contact, the first held by the stub +agent until the second has arrived, produce two turns, and the second turn's +prompt carries both messages' text. + +## README + +`README.md`, section "Connect an agent", states today that "A second message for +that conversation waits for the turn in progress". Rewrite that paragraph, in the +file's voice — plain statements, one idea per sentence: + +- One turn runs at a time for each conversation, and one conversation is one + contact on one channel. +- A conversation is busy until the reply of its turn has been delivered in full, + the paced pauses included. +- Messages that arrive while it is busy are sent together as the prompt of the + next turn, in the order they arrived, so a contact who writes three lines gets + one answer, and the agent starts a turn only after the contact has received + everything it said before. +- A message for another conversation does not wait, and the same contact id on two + channels is two conversations. +- A turn that reaches `turn_timeout` fails, the subprocess is stopped, and the + paragraphs it already produced are still delivered; the messages waiting behind + it are then sent. + +The paragraph on delivery states today that a message which cannot be sent ends +the turn's delivery. Add that a turn which was stopped for any other reason +delivers what it already produced, so what the contact received and what the +agent's session holds are the same. + +## Out of scope + +- Steering: cancelling the turn in flight when a message arrives, and the + distinction between a superseded turn and a failed one (PR 5b). +- Retrying a turn whose subprocess died (PR 5c). +- Waiting for in-flight turns and their queued paragraphs at shutdown, and taking + a message's position in `respondio.ServeHTTP` before the ack (PR 5d). Until + then, two webhooks that arrive in the same instant can enter the batch in either + order, because `respondio` hands each one to a goroutine of its own and the + scheduler decides which reaches the core first. +- Generating the next turn while the previous reply is still being paced out. It + keeps the harness busy, and it lets the delivery fall behind without a bound, so + the agent answers messages the contact wrote before reading the reply before + them. Holding the turn keeps the agent and the contact in step. +- Any setting for merging, batch size or queue depth. +- Deduplicating a message respond.io delivers twice. + +## References + +- Agent Client Protocol v1, + [Prompt Turn](https://agentclientprotocol.com/protocol/v1/prompt-turn) (checked + 2026-06): one prompt is in flight per session, and the `session/prompt` response + follows the updates of its turn. This is why a waiting caller must wait for the + turn that carries its content instead of returning early. diff --git a/PR-7-IMPLEMENTATION-PLAN.md b/PR-7-IMPLEMENTATION-PLAN.md new file mode 100644 index 0000000..34c7d6f --- /dev/null +++ b/PR-7-IMPLEMENTATION-PLAN.md @@ -0,0 +1,26 @@ +# PR 7 implementation plan + +1. **Extend ACP metadata in the core shape** + - Add embedded-resource MIME type and resource-link size. + - Add ACP pass-through tests before production changes. + - Do not add file storage or respond.io attachment behavior. + +2. **Add qualified conversation file storage** + - Add the minimal core contract. + - Implement safe stream storage in the agent conversation directory. + - Qualify conversation names in channel wiring and pass the contract to factories. + - Do not download respond.io attachments. + +3. **Receive respond.io attachments** + - Decode attachment events. + - Download HTTP(S) files with redirect, size, cleanup, filename, and failure controls. + - Acknowledge valid webhooks before downloads. + - Do not send outbound attachments. + +4. **Send respond.io attachments** + - Convert typed ACP blocks and classified text URLs to ordered respond.io calls. + - Stop at the first invalid block or failed call. + - Do not add upload or hosting for data-only blocks. + +5. **Document attachment transport** + - Document settings, storage, accepted schemes, failures, outbound ordering, and data-only limits. diff --git a/PR-7-REQUIREMENTS.md b/PR-7-REQUIREMENTS.md new file mode 100644 index 0000000..874d750 --- /dev/null +++ b/PR-7-REQUIREMENTS.md @@ -0,0 +1,33 @@ +# PR 7: respond.io attachment transport + +Carry attachments between respond.io and the agent without changing ACP channel behavior. + +- Keep ACP v1 content blocks as the one internal content shape. Extend the existing core types only for ACP media and resource metadata that this work needs. Do not add respond.io content types to the core. +- Keep the ACP channel as pass-through and preserve each ACP v1 shape: an image has data, MIME type, and an optional URI; audio has data and MIME type; an embedded resource has its URI and optional MIME type inside `resource`; and a resource link has URI, name, optional MIME type, and optional size. +- Add only the missing ACP fields to the shared core shape: embedded-resource MIME type and resource-link size. Do not add fields to content types that ACP v1 does not define. +- Convert only at the respond.io edge. A respond.io attachment maps to an ACP resource link, and an outbound ACP block with a usable public URL maps to a respond.io `attachment` message with type `image`, `video`, `audio`, or `file`. +- Put the smallest conversation-file contract in `internal/core`. The contract saves a named stream for a conversation and returns its safe name and URI. The agent-side conversation directory owner implements it, `channel.Build` qualifies the conversation name before use, and channel factories receive the contract. respond.io must not import agent packages, read `data_dir`, call harness methods, or build agent directory paths. +- Save every attachment from an incoming respond.io `message.received` event that Momo owns in that conversation's directory before the agent turn starts. The agent receives one ACP resource link for each saved file. +- Accept attachment download URLs only with an `http` or `https` scheme. Apply the same rule to every redirect target. +- Follow redirects without resetting the per-file byte limit. Reject a declared `Content-Length` above the limit before writing, stop a stream when it exceeds the limit, and count the bytes from the response body rather than trusting the declared size. +- Add `max_attachment_bytes` to the respond.io channel settings. Its default is `20,000,000` decimal bytes per file. Reject a configured value that is not positive. +- Keep the existing webhook body limit separate from `max_attachment_bytes`. The webhook handler must verify and acknowledge a valid event before it downloads an attachment. +- Write each download to a temporary file in the target conversation directory. Rename it only after a complete download within the limit. Remove the temporary file after an HTTP error, redirect error, read error, size error, write error, or cancelled download. +- Treat every remote filename as untrusted. Remove directory parts, control characters, empty names, `.` and `..`. Keep the final path inside the conversation directory, do not follow a file path out through a symlink, and do not overwrite an existing file. Generate a unique safe name when the proposed name is unsafe, empty, or already used. +- Set the resource link name to the saved basename. Use provider name and MIME metadata when the webhook supplies them, then HTTP `Content-Disposition` and `Content-Type`, then the final URL path and its extension. Use `application/octet-stream` when no MIME type is known. Include the stored byte count in the ACP resource link. +- If an attachment URL is absent or invalid, the download fails, the response is not successful, the file is too large, or storage fails, still run the owned conversation's agent turn. Replace that attachment with a plain ACP text block that says the attachment is unavailable and gives its safe display name and a short reason. +- Errors should be surfaced to the channel as usual. +- Map an outbound ACP image with an HTTP(S) URI by its image MIME type. Map ACP resource links by their MIME type and embedded resources by `resource.mimeType`: `image/*`, `video/*`, and `audio/*` use those attachment types, and every other MIME type uses `file`. +- If an outbound block has no MIME type, classify its HTTP(S) URI from the case-insensitive extension in the parsed URL path. Ignore query text and fragments. Use Go's known MIME extension table. A known non-media MIME type maps to `file`. +- A typed media or resource block without a usable HTTP(S) URI cannot form a respond.io attachment payload. This includes every ACP audio block because ACP v1 audio has no URI. Return a delivery error instead of dropping the block or sending base64 data as a URL. Keep upload or hosting of data-only ACP media out of PR 7. +- Find absolute HTTP(S) URLs in outbound ACP text blocks. Classify each URL only from its parsed path extension with the same MIME rules. Never request a plain text URL to inspect its headers or content. Keep a URL with no known extension as text. +- Replace each classified media or file URL in mixed text with one attachment send at that position. Send non-empty text before and after it as text messages. Preserve block order and URL order, and do not include adjacent punctuation in the URL. +- Make one respond.io API call for each ordered text or attachment part. Use `{ "message": { "type": "attachment", "attachment": { "type": "", "url": "" } } }` for an attachment. Do not start the next call until the current call succeeds. +- Stop an outbound sequence at its first invalid typed block or failed respond.io API call. Return the error so the existing turn failure report adds one internal comment. Do not retry or roll back parts that respond.io already accepted. +- Write tests before production code. Start with failing cases for inbound image, video, audio, and file payloads, ACP resource-link metadata, safe and duplicate filenames, unsupported URL schemes, redirects, declared and streamed size limits, partial-file cleanup, download failures and unavailable markers, webhook acknowledgement before download, typed outbound blocks, mixed text with several URLs, query strings and punctuation, unknown extensions left as text, ordered sends, and stop-on-first-failure behavior. +- Add pass-through tests for ACP media and resource metadata. Add contract and wiring tests that prove respond.io uses a qualified conversation file store without access to agent or configuration internals. +- Update the README configuration sample and respond.io guide with `max_attachment_bytes`, its decimal-byte default, saved-file location, accepted URL schemes, inbound failure behavior, outbound URL classification, ordering, and the limits on data-only blocks. +- Files likely touched: `internal/core/core.go`, a new `internal/core` conversation-file contract and tests, `internal/agent/agent.go`, `internal/agent/harness.go`, and their tests for safe conversation storage, `internal/channel/channel.go` and `internal/channel/channel_test.go` for factory wiring and qualification, `internal/channel/acp/acp.go`, `internal/channel/acp/methods.go`, and ACP tests for the shared contract and metadata pass-through, `internal/channel/respondio/respondio.go`, `internal/channel/respondio/client.go`, and their tests, `cmd/momo/main.go` and `cmd/momo/main_test.go` for wiring, and `README.md`. `internal/config` should not need a source change because respond.io owns and strictly decodes its settings. +- Exclude attachment upload or hosting, data-only ACP media conversion, content sniffing, malware scanning, retention policy, total conversation storage quotas, webhook body-limit changes, changes to ACP transport behavior, and media context for contact, operator, or workflow messages while another assignee owns the conversation. +- Base implementation only on this file and the current tracked source. Do not use closed pull requests, deleted work, git history, or the stale respond.io-only file list in `PR-PLAN.md`. +- Official references checked 2026-04-06: ACP [content](https://agentclientprotocol.com/protocol/v1/content) and [prompt turns](https://agentclientprotocol.com/protocol/v1/prompt-turn), plus respond.io [webhook overview](https://developers.respond.io/docs/webhooks/lwocem4sq812x-overview), [incoming message](https://developers.respond.io/docs/webhooks/5edb782537996-new-incoming-message), [outgoing message](https://developers.respond.io/docs/webhooks/5f71ee4218268-new-outgoing-message), and [send a message](https://developers.respond.io/docs/api/a748f5bfb1bb5-send-a-message). diff --git a/PR-PLAN.md b/PR-PLAN.md new file mode 100644 index 0000000..3dccb63 --- /dev/null +++ b/PR-PLAN.md @@ -0,0 +1,144 @@ +# PR plan: splitting feat/agent-server (#1) into incremental PRs + +The full implementation lives on `feat/agent-server` (PR #1, ~1,900 lines). It splits into a sequence of PRs, ordered by dependency: each one compiles, passes tests, and does something demonstrable end-to-end, and each introduces one new concept cluster — both architecturally and Go-wise. Content is re-authored from the branch tip (copy files, trim forward references), not extracted by rebasing. + +Workflow: branch each PR off the previous one, merge bottom-up into `main`. Tests grow with the stack — each PR proves itself. + +## PR 1 — Webhook intake ✅ (opened as #2, branch `feat/webhook-intake`) + +HTTP skeleton, config, webhook parsing and signature verification. + +- Files: `main.go`, `config.go`, `server.go` (routing half), `server_test.go`, `channel/channel.go` (domain contract: `channel.Message`, `channel.Handler`), `channel/respondio/` (webhook parsing, signature verification) +- Server boots, verifies per-webhook HMAC-SHA256 signatures, dispatches on `event_type`, acks immediately, handles async. Handlers are logging stubs. +- Channel abstraction (from PR review): core defines the channel-neutral vocabulary and never imports respond.io types or `net/http` in the domain package; the respondio package translates webhooks into `channel.Message`s and calls the core `Handler`. No lifecycle interface or registry until a second channel exists. +- Deliberately excluded: `assignee`/`attachment` payload fields (used only by later PRs), and the ACP dependency. +- Go concepts: project layout, `net/http`, env config, structs/JSON tags, graceful shutdown, `httptest`. + +## PR 2 — respond.io client, replies close the loop + +Wire a trivial responder (echo) so the webhook → REST API round trip works end-to-end. + +- Files: `channel/respondio/` grows the REST client; `channel/channel.go` gains the `Sender` interface (`SendText`); `server.go` grows a real `Incoming`; fake respond.io API scaffolding in tests (`fakeRespond`, `setupServer`, `wait`) +- `sendText` via `POST /contact/id:{id}/message` with bearer auth; `RESPOND_API_TOKEN` and `RESPOND_API_URL` config, parsed by the respondio package. +- Go concepts: HTTP client, error wrapping, test doubles. + +## PR 3 — ACP harness: spawn, session, one blocking turn + +The real agent replaces the echo. Biggest conceptual jump (subprocesses, stdio JSON-RPC, ACP lifecycle) — keep everything else out. + +- Introduce the harness abstraction alongside the first implementation, mirroring the channel/respond.io split from PR 1's review: a small domain contract (run a turn for a contact — prompt in, reply out) defined by the core, with the ACP subprocess harness as the first implementation in its own package. The core must not import ACP types or know about subprocesses/stdio. +- Spawn the harness per contact with cwd `DATA_DIR/` (seeded from `CONTACT_TEMPLATE` symlinks), `initialize` → `session/list`/`load`/`new` → `prompt`, deliver the full reply as one message, SIGTERM after the turn. +- Placeholder: a naive per-contact mutex serializes turns (replaced in PR 5). Auto-approve permission requests. +- The stub `testagent` lands here; the e2e test starts exercising webhook → prompt → reply. +- Go concepts: `os/exec`, pipes, JSON-RPC over stdio, interfaces (`acp.Client`). + +## PR 4 — Human-paced chunked delivery + +Switch from "send the final text" to streaming. + +- Files: `delivery.go`, `delivery_test.go`; `acp.go` routes `agent_message_chunk` updates into the active `turn` +- Paragraph splitting on `\n\n`, per-paragraph typing delay (`TYPING_DELAY_MS_PER_WORD` × words), ordered delivery via a queue goroutine. +- Go concepts: goroutines, channels, mutexes — isolated in one small, well-tested file. + +## PR 5 — One turn at a time for one conversation + +Four pull requests, in this order. Each caller of a turn waits for its reply — the +ACP channel answers `session/prompt` after the updates, and respond.io reports a +failed turn — so the turns of one conversation are ordered by a lock the callers +queue on, not by a goroutine and an inbox: a mailbox that hands every message its +result back is machinery for nothing. + +### 5a — Serialize the turns and merge what arrives during one + +- Files: `internal/core/conversation.go`, `internal/core/conversation_test.go`; + `internal/core/delivery.go`; `internal/agent/harness.go` loses `acquire`, its map + and its mutex. Written up in `PR-5A-REQUIREMENTS.md`. +- The rule is about conversations, not about ACP subprocesses, and a second agent + implementation would need it again, so it arrives as a `Handler` decorator in the + core, wired by `channel.Build` outside the delivery, and is tested against a fake + handler with no subprocess. +- A conversation is busy from the arrival of a message until the last paragraph of + the reply has been sent. Messages that arrive meanwhile accumulate in one pending + batch and reach the agent as a single prompt: a contact who writes three lines + gets one answer, no queue depth needs a bound because there is no queue, and a + turn starts only after the contact received everything the session history holds. +- A part that is queued for delivery is delivered even when the turn's context is + cancelled, so the contact's view and the agent's session never diverge. Only a + failed send drops what is behind it. +- The conversation's state is dropped when it is free and nothing is pending, under + the lock an arriving message takes to find it, so the map stops growing with the + contact base. +- Excluded: cancelling the turn in flight, and generating a turn while the previous + reply is still being paced out. +- Go concepts: `sync.Mutex` invariants, a decorator over an interface, + `context.WithoutCancel`. + +### 5b — Steering + +- ACP v1 carries one prompt in flight per session, offers `session/cancel` as the + only interruption, and has the agent answer the cancelled prompt with the + `cancelled` stop reason before the client prompts again + ([Prompt Turn](https://agentclientprotocol.com/protocol/v1/prompt-turn)). A + second prompt on a running session is outside the protocol, and pi's own + non-destructive `steer` is reachable only through pi's RPC mode. So momo steers + with the protocol: cancel, wait for the cancelled response, then prompt again on + the same session in the same subprocess, carrying the interrupted message and + everything that arrived after it. +- The stop is not immediate. The agent cuts the model stream at once, ends the tool + call it is running, and only then ends the turn. Everything that arrives in that + window merges into the next prompt, which is what 5a already does. +- A superseded turn is not a failed turn. The core says the difference, respond.io + posts no failure comment for it, and the ACP channel answers `cancelled` instead + of an internal error. +- The agent keeps its own history valid: pi writes a tool result for every call the + interrupt hit and keeps the partial assistant message. momo constructs nothing. +- Files: `internal/core/conversation.go`, `internal/agent/`, both channels. +- Go concepts: context cancellation, ACP cancellation. + +### 5c — Retry a turn whose harness died + +- One retry when the subprocess dies mid-turn. `internal/agent/` only, and + independent of ordering. + +### 5d — respond.io holds its turns for the shutdown, and takes their order + +- `ServeHTTP` acks and runs the turn on a goroutine with `context.WithoutCancel`, + so a shutdown neither stops it nor waits for it. The channel's lifetime stops new + work and a wait group drains what runs. +- The same handler is where the conversation's position is taken, before the ack: a + goroutine started per webhook reaches the core in the order the scheduler runs + it, and two webhooks that arrive in the same instant can swap. +- Files: `internal/channel/respondio/`. + +## PR 6 — Human handover through session history + +Let a human operator control a respond.io conversation while the agent session keeps +what the contact and operator write. + +## PR 7 — Attachments + +Carry media through the shared ACP content shape and convert it only at the +respond.io boundary. + +- Inbound: save attachments in the conversation directory and send ACP resource links to the agent. +- Outbound: send typed ACP media and media URLs in mixed text as respond.io attachments. +- Apply safe filenames, ordered delivery, and a configurable 20 MB download limit. +- Files: `internal/core/`, `internal/agent/`, `internal/channel/`, `cmd/momo/`. + +## PR 8 — Conversation template files + +Give the harness a scoped context in every conversation directory: instructions (`AGENTS.md`), skills, and any other file the operator wants. The operator configures one file source. momo copies its files into the working directory before each turn, so a resumed conversation never runs on stale files. + +- Concept: a template source is the first place in momo where a variation point is real (local directory, S3 bucket, GitHub repository). Follow the channel seam: the core owns the contract and the registry, each source lives in its own package, decodes its own settings block, and registers itself. The core asks the registry what exists; nothing enumerates the sources. +- Contract: a source lists its files and reads their content with a version marker (mtime and size for local, ETag for S3, commit or blob sha for GitHub). Momo compares the marker with what the directory holds and writes only the files that differ. +- Sync runs before every turn, not only at start-up, because a conversation resumes days after it began and the template changed in the meantime. A file that the harness itself created stays; only template paths are managed. +- One source in the first PR (local directory), because it needs no network and it proves the seam. S3 and GitHub are then purely additive: one new package each, one import in `cmd/momo`, no edit to the contract. Split them into PR 8b and PR 8c. +- Files: template contract and sync in the core side, `internal/template/local/`, settings block, tests with a temporary directory. +- Tests: files appear in the working directory before the prompt; a changed template file is rewritten on the next turn; an unchanged file is not rewritten; a file that the harness wrote is left alone; a missing or unreadable source stops the process at start-up. +- Go concepts: `io/fs`, `os.WriteFile`, content comparison, table tests with `t.TempDir()`. + +## Notes + +- PRs 3–4 briefly contain placeholder behaviour (blocking turn, naive locking) that PR 5a deletes — flag it in the PR description ("replaced in the next PR") so reviewers don't polish throwaway code. +- Each PR is roughly 150–350 lines. +- When PR 4/5 settles the delivery concurrency shape, revisit the respond.io client's transport: `DefaultTransport` caps `MaxIdleConnsPerHost` at 2, so more than 2 concurrent sends to the API pay fresh TCP+TLS handshakes instead of reusing keep-alive connections.