From 395eee5ed41a73105d20171d867a32f661d42b3e Mon Sep 17 00:00:00 2001 From: Josh Rotenberg Date: Mon, 12 Oct 2026 08:12:13 -0700 Subject: [PATCH 1/2] feat: plan operator attachment references (#846) From 915c92535b69b34badb65234574ab5f2715c972a Mon Sep 17 00:00:00 2001 From: Josh Rotenberg Date: Mon, 12 Oct 2026 08:33:30 -0700 Subject: [PATCH 2/2] feat: expose durable operator image attachments (#846) --- docs/mcp-reference.json | 204 +++++++- docs/mcp-reference.md | 209 +++++++- docs/mcp/behavior.json | 24 +- docs/mcp/overview.md | 47 ++ lib/custode/janitor.ex | 33 +- lib/custode/mcp/attachment_tools.ex | 118 +++++ lib/custode/mcp/server.ex | 2 + lib/custode/mcp/snodo.ex | 1 + lib/custode/mcp/tool_policy.ex | 2 + lib/custode/mcp/tools.ex | 24 +- lib/custode/operator/actions.ex | 12 +- lib/custode/operator/attachment.ex | 32 ++ lib/custode/operator/attachments.ex | 484 ++++++++++++++++-- lib/custode/operator_message.ex | 2 + lib/custode/operator_messages.ex | 114 +++-- lib/custode/uploads.ex | 11 +- lib/custode_web/console_live.ex | 47 +- lib/custode_web/conversation_live.ex | 45 +- ...0261012151227_add_operator_attachments.exs | 33 ++ test/custode/attachments_test.exs | 27 +- test/custode/janitor_test.exs | 42 ++ test/custode/mcp_attachment_test.exs | 425 +++++++++++++++ .../operator_attachment_messages_test.exs | 314 ++++++++++++ .../operator_attachment_storage_test.exs | 330 ++++++++++++ test/custode_web/console_live_test.exs | 44 ++ test/custode_web/conversation_live_test.exs | 60 +++ test/custode_web/feed_thumbnails_test.exs | 13 + test/test_helper.exs | 1 + 28 files changed, 2549 insertions(+), 151 deletions(-) create mode 100644 lib/custode/mcp/attachment_tools.ex create mode 100644 lib/custode/operator/attachment.ex create mode 100644 priv/repo/migrations/20261012151227_add_operator_attachments.exs create mode 100644 test/custode/mcp_attachment_test.exs create mode 100644 test/custode/operator_attachment_messages_test.exs create mode 100644 test/custode/operator_attachment_storage_test.exs diff --git a/docs/mcp-reference.json b/docs/mcp-reference.json index 5a9f11e7..6ed4402e 100644 --- a/docs/mcp-reference.json +++ b/docs/mcp-reference.json @@ -45,6 +45,8 @@ "approve_action", "ask_operator", "assurance_read", + "attachment_read", + "attachment_upload", "await_agent", "beat", "compact_journal", @@ -848,6 +850,187 @@ ], "name": "assurance_read" }, + { + "behavior": { + "access": "Main endpoint, verified authenticated human operator with nonempty identity ID only. The caller must own the reference. No fallback caller; routine and subagent tokens refuse.", + "effects": "Reads retained bytes after ownership, current workspace, regular-file, digest and retention checks. Does not create a message, start work or extend retention.", + "notes": "Required attachment_id is 1 to 128 characters. Unknown, foreign, expired, moved-workspace, missing or changed image references refuse without substituting other bytes. Historical message references may remain after image retention ends. Reading an image does not prove that an agent read or understood it.", + "result": "Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention, ISO created_at and data_base64. No private path is returned.", + "summary": "Read owned image metadata and Base64 bytes by an opaque reference." + }, + "category": "operator", + "definition": { + "annotations": {}, + "description": "Read an image by its server-owned attachment reference as the same verified human operator who uploaded it. Returns bounded metadata and Base64 bytes, never a selectable filesystem path. Reading does not start work.", + "inputSchema": { + "additionalProperties": false, + "properties": { + "attachment_id": { + "maxLength": 128, + "minLength": 1, + "type": "string" + } + }, + "required": [ + "attachment_id" + ], + "type": "object" + }, + "name": "attachment_read", + "outputSchema": { + "additionalProperties": false, + "properties": { + "agent_id": { + "type": "string" + }, + "attachment_id": { + "type": "string" + }, + "byte_count": { + "maximum": 10000000, + "minimum": 1, + "type": "integer" + }, + "created_at": { + "type": "string" + }, + "data_base64": { + "maxLength": 13333336, + "type": "string" + }, + "media_type": { + "enum": [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp" + ], + "type": "string" + }, + "retention": { + "type": "string" + }, + "sha256": { + "maxLength": 64, + "minLength": 64, + "type": "string" + } + }, + "required": [ + "agent_id", + "attachment_id", + "byte_count", + "created_at", + "data_base64", + "media_type", + "retention", + "sha256" + ], + "type": "object" + }, + "title": "attachment_read" + }, + "endpoints": [ + "/mcp" + ], + "name": "attachment_read" + }, + { + "behavior": { + "access": "Main endpoint, verified authenticated human operator with nonempty identity ID only. No fallback caller. Shared storage repeats the identity check and binds the reference to this operator and configured routine.", + "effects": "Validates the declared media type, decoded size and image header, then stores bytes under a server-owned path in the routine workspace and records ownership. Does not create a message, start an agent or call a model.", + "notes": "Required agent_id, media_type and data_base64. Types are image/png, image/jpeg, image/gif and image/webp. Decoded bytes must be at most 10,000,000; Base64 length at most 13,333,336. Validation checks the header, not full codec validity. Clients cannot choose paths. Identical live bytes for the same operator, routine and workspace reuse the retained reference. Active messages pin their attachment; otherwise configured upload retention applies. Janitor commits an expired reference before attempting file cleanup; reads then refuse it. Failed file removals are retried on a later prune. Main MCP requests allow 14,000,000 bytes to accommodate the image envelope; the memory endpoint keeps its 2,000,000-byte limit. Errors omit private paths and raw exceptions.", + "result": "Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention and ISO created_at. No path or image bytes in this result.", + "summary": "Store one bounded image and return an owned reference for an operator message." + }, + "category": "operator", + "definition": { + "annotations": {}, + "description": "Upload one bounded PNG, JPEG, GIF or WebP image for a configured routine as its verified human operator. Returns an owned attachment reference for prompt_agent; uploading does not start work. The server validates size and image header, not full codec validity.", + "inputSchema": { + "additionalProperties": false, + "properties": { + "agent_id": { + "maxLength": 1024, + "minLength": 1, + "type": "string" + }, + "data_base64": { + "maxLength": 13333336, + "minLength": 1, + "type": "string" + }, + "media_type": { + "enum": [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp" + ], + "type": "string" + } + }, + "required": [ + "agent_id", + "data_base64", + "media_type" + ], + "type": "object" + }, + "name": "attachment_upload", + "outputSchema": { + "additionalProperties": false, + "properties": { + "agent_id": { + "type": "string" + }, + "attachment_id": { + "type": "string" + }, + "byte_count": { + "maximum": 10000000, + "minimum": 1, + "type": "integer" + }, + "created_at": { + "type": "string" + }, + "media_type": { + "enum": [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp" + ], + "type": "string" + }, + "retention": { + "type": "string" + }, + "sha256": { + "maxLength": 64, + "minLength": 64, + "type": "string" + } + }, + "required": [ + "agent_id", + "attachment_id", + "byte_count", + "created_at", + "media_type", + "retention", + "sha256" + ], + "type": "object" + }, + "title": "attachment_upload" + }, + "endpoints": [ + "/mcp" + ], + "name": "attachment_upload" + }, { "behavior": { "access": "Main endpoint capability: operator, caretaker, or specialist routine. A routine may await only a temporary agent whose durable spawn record names it as parent; the operator may await any target. Missing or reconciled records do not widen access.", @@ -1778,7 +1961,7 @@ "access": "Main endpoint capability: human operator only. The tool is named in no routine's exposure list, so only the operator discovers it, and the handler repeats the human-only check.", "effects": "None; reads existing durable operator-message records.", "notes": "Returns only messages authored by the authenticated operator identity, newest first. agent_id and status are optional exact filters. status accepts queued, executing, waiting_for_input, waiting_for_approval, completed, failed or refused. limit defaults to 20, must be positive and is capped at 100. Each row carries only a 160-character prompt preview for disambiguation; full prompts, idempotency keys and hashes, claim fields, results, errors and provider-turn/session details are omitted. Use the recovered agent_id and message_id with await_agent for the authoritative exact receipt.", - "result": "messages array of message_id, agent_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps.", + "result": "messages array of message_id, agent_id, nullable attachment_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps. attachment_id retains the original image reference; it is null for messages without a managed attachment.", "summary": "Discover this operator's durable messages after a fresh session or reconnect." }, "category": "read", @@ -2978,11 +3161,11 @@ }, { "behavior": { - "access": "Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access.", + "access": "Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access. Supplying attachment_id requires a verified authenticated human identity with a nonempty ID; routine and subagent delegation refuses attachments explicitly rather than dropping them. The attachment path never uses the legacy default caller identity.", "effects": "Creates a durable operator-message record, then sends or queues work exactly once for a new caller/target/idempotency tuple. For an operator, can resume a paused agent or start an offline configured routine and records activity. If a configured routine is adopting a changed provider contract, the durable message waits behind that boundary and is submitted after the replacement is ready.", - "notes": "agent_id and prompt are required. idempotency_key is optional; omission creates a new logical message, while retrying the same caller, target and key with the same prompt returns the existing receipt and duplicate=true. Reusing the key with different text is a tool error. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent.", - "result": "The durable message receipt: message_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate.", - "summary": "Send one durable, idempotent message or answer to an agent." + "notes": "agent_id and prompt are required. Optional attachment_id is a 1 to 128 character string owned by the operator and bound to this routine; one image per message. Empty prompt with a valid image is supported. Explicit attachment_id:null is invalid; omit it for text-only messages. Ownership, target, workspace and retained bytes are validated during atomic message admission; a failed image refuses the whole message. idempotency_key is optional; omission or null creates a new logical message, while retrying the same caller, target and key with identical text and attachment reference returns the existing receipt and duplicate=true. Changing either text or image reference is a tool error. A valid existing receipt remains retrievable after image retention expires; it does not restore expired bytes. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent.", + "result": "The durable message receipt: message_id, nullable attachment_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate. attachment_id is the originally admitted reference, not an assertion of image consumption by the model.", + "summary": "Send one durable, idempotent message or answer to an agent, optionally with an owned image reference." }, "category": "delegate", "definition": { @@ -2993,9 +3176,18 @@ "agent_id": { "type": "string" }, + "attachment_id": { + "description": "one image reference owned by the verified operator and bound to this routine", + "maxLength": 128, + "minLength": 1, + "type": "string" + }, "idempotency_key": { "description": "deduplicate this caller's delivery to this agent", - "type": "string" + "type": [ + "string", + "null" + ] }, "prompt": { "type": "string" diff --git a/docs/mcp-reference.md b/docs/mcp-reference.md index b774af43..2dc4de23 100644 --- a/docs/mcp-reference.md +++ b/docs/mcp-reference.md @@ -85,11 +85,58 @@ Resource URI variables are required opaque strings. Percent-encode each path seg Routine instructions, prompt files, the `prompt_agent` tool, and an agent's structured permission requests are different mechanisms from MCP prompts. Consult the generated Prompts section for registered MCP prompt templates and their arguments. + +## Operator image attachments + +An authenticated human client can send one PNG, JPEG, GIF or WebP image with a +message, including an image without text: + +1. Call `attachment_upload` with `agent_id`, `media_type` and `data_base64`. +2. Retain the returned `attachment_id`. +3. Call `prompt_agent` with the same `agent_id`, that `attachment_id`, a stable + `idempotency_key`, and `prompt` (use an empty string for image-only messages). +4. Follow the returned `message_id` with `await_agent`. The receipt includes its + `attachment_id`; acceptance does not mean the provider has read the image. + +Uploads and reads are human-only. A reference belongs to the verified operator +identity and the target routine, regardless of client or transport. Knowing a +reference does not let a different operator read or submit it, and routine and +helper identities cannot use attachments. No upload or retrieval argument names +a server filesystem path. `attachment_read` returns metadata and base64 bytes to +the owning operator. The dashboard's existing authenticated image routes remain +available to the local operator. + +The shared UI/MCP limit is 10,000,000 decoded bytes per image. The main MCP HTTP +endpoint accepts at most 14,000,000 body bytes to accommodate base64 and JSON; +the memory endpoint remains limited to 2,000,000 bytes. Upload validation checks +size and recognized image headers against the declared type. It does not decode +or sanitize the complete image, or establish that a downstream decoder will +accept it. JPEG uploads use the canonical `image/jpeg` type. + +Identical live uploads by the same operator to the same routine/workspace return +the same reference. Uploading alone does not start an agent. Prompt submission +uses the existing start/resume/queue operation, atomically retaining the image +reference with the message before provider admission. Its idempotency fingerprint +uses the submitted text and reference, not a workspace path. Retrying an accepted +message with the same key and input returns its retained receipt even after bytes +expire or the routine moves; it does not start another turn. A changed text or +reference under that key conflicts. New submissions require available bytes in +the current routine workspace. + +Cleanup follows the configured upload retention (30 days by default). Abandoned +uploads age from upload time. Queued, executing, waiting-for-input and +waiting-for-approval messages protect their image regardless of age. After the +last referencing message settles, retention runs from its last update. Cleanup +keeps reference metadata and message receipts after removing bytes, so old +receipts remain meaningful but retrieving an expired image fails. Disabling +upload retention disables this cleanup. This is an age policy, not a storage +quota or a promise that bytes survive manual filesystem changes. + ## Registered endpoint inventory | Endpoint | Server | Tools | Resources | Templates | Prompts | | --- | --- | ---: | ---: | ---: | ---: | -| `/mcp` | custode 0.3.0 | 107 | 4 | 13 | 0 | +| `/mcp` | custode 0.3.0 | 109 | 4 | 13 | 0 | | `/mcp/memory` | memory 0.3.0 | 7 | 0 | 0 | 0 | **`/mcp`**: protocol versions 2026-07-28, 2025-11-25, 2025-06-18; capabilities `{"resources":{},"tools":{}}`. @@ -112,6 +159,8 @@ Categories are descriptive policy metadata, not an authorization guarantee. | [approve_action](#tool-approve_action) | delegate | | [ask_operator](#tool-ask_operator) | self_write | | [assurance_read](#tool-assurance_read) | read | +| [attachment_read](#tool-attachment_read) | operator | +| [attachment_upload](#tool-attachment_upload) | operator | | [await_agent](#tool-await_agent) | read | | [beat](#tool-beat) | operator | | [compact_journal](#tool-compact_journal) | self_write | @@ -367,6 +416,151 @@ Read exact scoped assurance predicates and retained evidence and decision refere **Behavior, defaults and errors:** Opt-in assurance_assignments configuration defaults empty. One configured owner and designated human judge, frozen objective/input/criteria/artifact/policy and up to16 revision rounds; up to256 retained events. The shared recorder accepts exact existing source references, never caller trust classes or source payloads. Existing OwnerReviews are self-reported opinions; absent native run identity cannot satisfy independent predicates. Document publication/current bytes prove presence only. Existing repository check projections omit issuer and check head, so their propositions remain unknown. Read evaluates current policy; changed policy invalidates acceptance and old receipts never become current by relabeling. Historical decisions retain the exact evidence ids used. Accepted means all configured predicates and designated judge passed; it grants no effect authority. CLI assurance-read calls this same tool. Native source references bind recorder-owned provider/session/configuration and exact frozen command/cwd, artifact and check output; native_check reproduction and authored native_opinion remain distinct. Missing emitted identities/bindings remain unknown. Native raw output is retained once in the bounded recorder row, never copied into evidence snapshots. The native launcher is internal, explicit opt-in and unavailable through this read tool. Completed launch status does not attest all descendants settled; workspace reservations prevent automatic reuse or redelivery. Proof-harness judgment is synthetic and never actual human approval. Native check framing requires one exact bound JSON result, optionally preceded by the fixed macOS Git temp-directory diagnostic, which is retained. Ambiguous JSON, unrecognized diagnostics and inconsistent command terminals remain unknown; a native failed command with a nonzero exit can establish a failed independently reproduced check. +### Tool: attachment_read + +Read owned image metadata and Base64 bytes by an opaque reference. + +**Endpoints:** /mcp. **Category:** operator. + +| Argument | Type | Schema required | Description | Other schema constraints | +| --- | --- | --- | --- | --- | +| attachment_id | string | yes | | {"maxLength":128,"minLength":1} | + +**Result:** Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention, ISO created_at and data_base64. No private path is returned. + +**Side effects:** Reads retained bytes after ownership, current workspace, regular-file, digest and retention checks. Does not create a message, start work or extend retention. + +**Access:** Main endpoint, verified authenticated human operator with nonempty identity ID only. The caller must own the reference. No fallback caller; routine and subagent tokens refuse. + +**Behavior, defaults and errors:** Required attachment_id is 1 to 128 characters. Unknown, foreign, expired, moved-workspace, missing or changed image references refuse without substituting other bytes. Historical message references may remain after image retention ends. Reading an image does not prove that an agent read or understood it. + +**Output schema:** + +```json +{ + "additionalProperties": false, + "properties": { + "agent_id": { + "type": "string" + }, + "attachment_id": { + "type": "string" + }, + "byte_count": { + "maximum": 10000000, + "minimum": 1, + "type": "integer" + }, + "created_at": { + "type": "string" + }, + "data_base64": { + "maxLength": 13333336, + "type": "string" + }, + "media_type": { + "enum": [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp" + ], + "type": "string" + }, + "retention": { + "type": "string" + }, + "sha256": { + "maxLength": 64, + "minLength": 64, + "type": "string" + } + }, + "required": [ + "agent_id", + "attachment_id", + "byte_count", + "created_at", + "data_base64", + "media_type", + "retention", + "sha256" + ], + "type": "object" +} +``` + +### Tool: attachment_upload + +Store one bounded image and return an owned reference for an operator message. + +**Endpoints:** /mcp. **Category:** operator. + +| Argument | Type | Schema required | Description | Other schema constraints | +| --- | --- | --- | --- | --- | +| agent_id | string | yes | | {"maxLength":1024,"minLength":1} | +| data_base64 | string | yes | | {"maxLength":13333336,"minLength":1} | +| media_type | string | yes | | {"enum":["image/png","image/jpeg","image/gif","image/webp"]} | + +**Result:** Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention and ISO created_at. No path or image bytes in this result. + +**Side effects:** Validates the declared media type, decoded size and image header, then stores bytes under a server-owned path in the routine workspace and records ownership. Does not create a message, start an agent or call a model. + +**Access:** Main endpoint, verified authenticated human operator with nonempty identity ID only. No fallback caller. Shared storage repeats the identity check and binds the reference to this operator and configured routine. + +**Behavior, defaults and errors:** Required agent_id, media_type and data_base64. Types are image/png, image/jpeg, image/gif and image/webp. Decoded bytes must be at most 10,000,000; Base64 length at most 13,333,336. Validation checks the header, not full codec validity. Clients cannot choose paths. Identical live bytes for the same operator, routine and workspace reuse the retained reference. Active messages pin their attachment; otherwise configured upload retention applies. Janitor commits an expired reference before attempting file cleanup; reads then refuse it. Failed file removals are retried on a later prune. Main MCP requests allow 14,000,000 bytes to accommodate the image envelope; the memory endpoint keeps its 2,000,000-byte limit. Errors omit private paths and raw exceptions. + +**Output schema:** + +```json +{ + "additionalProperties": false, + "properties": { + "agent_id": { + "type": "string" + }, + "attachment_id": { + "type": "string" + }, + "byte_count": { + "maximum": 10000000, + "minimum": 1, + "type": "integer" + }, + "created_at": { + "type": "string" + }, + "media_type": { + "enum": [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp" + ], + "type": "string" + }, + "retention": { + "type": "string" + }, + "sha256": { + "maxLength": 64, + "minLength": 64, + "type": "string" + } + }, + "required": [ + "agent_id", + "attachment_id", + "byte_count", + "created_at", + "media_type", + "retention", + "sha256" + ], + "type": "object" +} +``` + ### Tool: await_agent Wait for one exact message, or use legacy agent-level waiting. @@ -859,7 +1053,7 @@ Discover this operator's durable messages after a fresh session or reconnect. | limit | integer | no | newest messages to return, default 20, capped at 100 | | | status | string | no | restrict to queued, executing, waiting_for_input, waiting_for_approval, completed, failed or refused | | -**Result:** messages array of message_id, agent_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps. +**Result:** messages array of message_id, agent_id, nullable attachment_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps. attachment_id retains the original image reference; it is null for messages without a managed attachment. **Side effects:** None; reads existing durable operator-message records. @@ -1349,23 +1543,24 @@ Read recent owner-authored interval reports, older open asks/gates and report fr ### Tool: prompt_agent -Send one durable, idempotent message or answer to an agent. +Send one durable, idempotent message or answer to an agent, optionally with an owned image reference. **Endpoints:** /mcp. **Category:** delegate. | Argument | Type | Schema required | Description | Other schema constraints | | --- | --- | --- | --- | --- | | agent_id | string | yes | | | -| idempotency_key | string | no | deduplicate this caller's delivery to this agent | | +| attachment_id | string | no | one image reference owned by the verified operator and bound to this routine | {"maxLength":128,"minLength":1} | +| idempotency_key | string, null | no | deduplicate this caller's delivery to this agent | | | prompt | string | yes | | | -**Result:** The durable message receipt: message_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate. +**Result:** The durable message receipt: message_id, nullable attachment_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate. attachment_id is the originally admitted reference, not an assertion of image consumption by the model. **Side effects:** Creates a durable operator-message record, then sends or queues work exactly once for a new caller/target/idempotency tuple. For an operator, can resume a paused agent or start an offline configured routine and records activity. If a configured routine is adopting a changed provider contract, the durable message waits behind that boundary and is submitted after the replacement is ready. -**Access:** Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access. +**Access:** Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access. Supplying attachment_id requires a verified authenticated human identity with a nonempty ID; routine and subagent delegation refuses attachments explicitly rather than dropping them. The attachment path never uses the legacy default caller identity. -**Behavior, defaults and errors:** agent_id and prompt are required. idempotency_key is optional; omission creates a new logical message, while retrying the same caller, target and key with the same prompt returns the existing receipt and duplicate=true. Reusing the key with different text is a tool error. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent. +**Behavior, defaults and errors:** agent_id and prompt are required. Optional attachment_id is a 1 to 128 character string owned by the operator and bound to this routine; one image per message. Empty prompt with a valid image is supported. Explicit attachment_id:null is invalid; omit it for text-only messages. Ownership, target, workspace and retained bytes are validated during atomic message admission; a failed image refuses the whole message. idempotency_key is optional; omission or null creates a new logical message, while retrying the same caller, target and key with identical text and attachment reference returns the existing receipt and duplicate=true. Changing either text or image reference is a tool error. A valid existing receipt remains retrievable after image retention expires; it does not restore expired bytes. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent. ### Tool: provision_owned_checkout diff --git a/docs/mcp/behavior.json b/docs/mcp/behavior.json index 98cab383..3510a527 100644 --- a/docs/mcp/behavior.json +++ b/docs/mcp/behavior.json @@ -107,6 +107,20 @@ } }, "tools": { + "attachment_upload": { + "summary": "Store one bounded image and return an owned reference for an operator message.", + "access": "Main endpoint, verified authenticated human operator with nonempty identity ID only. No fallback caller. Shared storage repeats the identity check and binds the reference to this operator and configured routine.", + "effects": "Validates the declared media type, decoded size and image header, then stores bytes under a server-owned path in the routine workspace and records ownership. Does not create a message, start an agent or call a model.", + "result": "Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention and ISO created_at. No path or image bytes in this result.", + "notes": "Required agent_id, media_type and data_base64. Types are image/png, image/jpeg, image/gif and image/webp. Decoded bytes must be at most 10,000,000; Base64 length at most 13,333,336. Validation checks the header, not full codec validity. Clients cannot choose paths. Identical live bytes for the same operator, routine and workspace reuse the retained reference. Active messages pin their attachment; otherwise configured upload retention applies. Janitor commits an expired reference before attempting file cleanup; reads then refuse it. Failed file removals are retried on a later prune. Main MCP requests allow 14,000,000 bytes to accommodate the image envelope; the memory endpoint keeps its 2,000,000-byte limit. Errors omit private paths and raw exceptions." + }, + "attachment_read": { + "summary": "Read owned image metadata and Base64 bytes by an opaque reference.", + "access": "Main endpoint, verified authenticated human operator with nonempty identity ID only. The caller must own the reference. No fallback caller; routine and subagent tokens refuse.", + "effects": "Reads retained bytes after ownership, current workspace, regular-file, digest and retention checks. Does not create a message, start work or extend retention.", + "result": "Closed structured attachment_id, agent_id, media_type, byte_count, sha256, retention, ISO created_at and data_base64. No private path is returned.", + "notes": "Required attachment_id is 1 to 128 characters. Unknown, foreign, expired, moved-workspace, missing or changed image references refuse without substituting other bytes. Historical message references may remain after image retention ends. Reading an image does not prove that an agent read or understood it." + }, "add_routine": { "access": "Main endpoint capability: operator or caretaker routine. A human operator may write directly; a caretaker must also have a live human-approved continuation whose action class is roster. Specialists and temporary agents are refused before configuration changes.", "effects": "Appends configuration, reloads the live roster and records a feed entry. The routine becomes beatable immediately and eligible for scheduling. Also mints credentials, writes the routine MCP config, initializes its workspace and starts repository service when configured.", @@ -300,7 +314,7 @@ "access": "Main endpoint capability: human operator only. The tool is named in no routine's exposure list, so only the operator discovers it, and the handler repeats the human-only check.", "effects": "None; reads existing durable operator-message records.", "notes": "Returns only messages authored by the authenticated operator identity, newest first. agent_id and status are optional exact filters. status accepts queued, executing, waiting_for_input, waiting_for_approval, completed, failed or refused. limit defaults to 20, must be positive and is capped at 100. Each row carries only a 160-character prompt preview for disambiguation; full prompts, idempotency keys and hashes, claim fields, results, errors and provider-turn/session details are omitted. Use the recovered agent_id and message_id with await_agent for the authoritative exact receipt.", - "result": "messages array of message_id, agent_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps.", + "result": "messages array of message_id, agent_id, nullable attachment_id, status, delivery, provider, continuation message id, prompt preview and inserted, started and completed timestamps. attachment_id retains the original image reference; it is null for messages without a managed attachment.", "summary": "Discover this operator's durable messages after a fresh session or reconnect." }, "list_policies": { @@ -444,11 +458,11 @@ "summary": "Provision a routine-owned clone at its deterministic Custode path." }, "prompt_agent": { - "access": "Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access.", + "access": "Main endpoint capability: operator, caretaker, or specialist routine. A routine may prompt only a temporary agent whose durable spawn record names it as parent; the operator may message any target through the shared operator action. Missing or reconciled records do not widen access. Supplying attachment_id requires a verified authenticated human identity with a nonempty ID; routine and subagent delegation refuses attachments explicitly rather than dropping them. The attachment path never uses the legacy default caller identity.", "effects": "Creates a durable operator-message record, then sends or queues work exactly once for a new caller/target/idempotency tuple. For an operator, can resume a paused agent or start an offline configured routine and records activity. If a configured routine is adopting a changed provider contract, the durable message waits behind that boundary and is submitted after the replacement is ready.", - "notes": "agent_id and prompt are required. idempotency_key is optional; omission creates a new logical message, while retrying the same caller, target and key with the same prompt returns the existing receipt and duplicate=true. Reusing the key with different text is a tool error. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent.", - "result": "The durable message receipt: message_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate.", - "summary": "Send one durable, idempotent message or answer to an agent." + "notes": "agent_id and prompt are required. Optional attachment_id is a 1 to 128 character string owned by the operator and bound to this routine; one image per message. Empty prompt with a valid image is supported. Explicit attachment_id:null is invalid; omit it for text-only messages. Ownership, target, workspace and retained bytes are validated during atomic message admission; a failed image refuses the whole message. idempotency_key is optional; omission or null creates a new logical message, while retrying the same caller, target and key with identical text and attachment reference returns the existing receipt and duplicate=true. Changing either text or image reference is a tool error. A valid existing receipt remains retrievable after image retention expires; it does not restore expired bytes. Waiting-for-user input creates a new public message linked to the question while retaining its provider correlation. Other compatible operator continuations restore the exact retained session handle even if the running provider evicted it from its bounded cache. A conflicting live handle is refused before enqueue; queued input is retained, never retried into a fresh session to bypass the conflict. how='queued' means the durable message is waiting because the provider is busy or because the routine is adopting its current execution configuration. how='admitting' means Custode has claimed the durable message for provider admission but has not yet durably recorded the final delivery path; use the returned message_id with await_agent. delivered=true means the durable submission was accepted rather than refused; it does not mean the provider turn completed. Operator how='started' means startup was accepted, not that execution began. A refusal before enqueue returns a durable refused receipt with delivered=false rather than losing the attempt. New clients should pass the returned message_id to await_agent.", + "result": "The durable message receipt: message_id, nullable attachment_id, target and caller provenance, status, delivery path, provider and exact turn identity when known, continuation link, detail, result, error and timestamps; plus agent_id, delivered, how and duplicate. attachment_id is the originally admitted reference, not an assertion of image consumption by the model.", + "summary": "Send one durable, idempotent message or answer to an agent, optionally with an owned image reference." }, "recall": { "access": "Main endpoint capability: operator, caretaker, or specialist routine; memory endpoint capability: temporary agent. Any authenticated caller reaching this tool may read another identity's memories by ID. No self-only read guard.", diff --git a/docs/mcp/overview.md b/docs/mcp/overview.md index 752401fd..dfc8a8db 100644 --- a/docs/mcp/overview.md +++ b/docs/mcp/overview.md @@ -78,3 +78,50 @@ the target server-side boundary. Resource URI variables are required opaque strings. Percent-encode each path segment and do not infer an ID format or construct cursors. Resource reads do not accept query parameters. List pages contain up to 25 items, a versioned `contract`, pagination metadata, and navigation links. Follow `page.next_uri` until null. Cursors are scoped to their collection; they are continuation handles, not a stable snapshot of changing data. Routine instructions, prompt files, the `prompt_agent` tool, and an agent's structured permission requests are different mechanisms from MCP prompts. Consult the generated Prompts section for registered MCP prompt templates and their arguments. + + +## Operator image attachments + +An authenticated human client can send one PNG, JPEG, GIF or WebP image with a +message, including an image without text: + +1. Call `attachment_upload` with `agent_id`, `media_type` and `data_base64`. +2. Retain the returned `attachment_id`. +3. Call `prompt_agent` with the same `agent_id`, that `attachment_id`, a stable + `idempotency_key`, and `prompt` (use an empty string for image-only messages). +4. Follow the returned `message_id` with `await_agent`. The receipt includes its + `attachment_id`; acceptance does not mean the provider has read the image. + +Uploads and reads are human-only. A reference belongs to the verified operator +identity and the target routine, regardless of client or transport. Knowing a +reference does not let a different operator read or submit it, and routine and +helper identities cannot use attachments. No upload or retrieval argument names +a server filesystem path. `attachment_read` returns metadata and base64 bytes to +the owning operator. The dashboard's existing authenticated image routes remain +available to the local operator. + +The shared UI/MCP limit is 10,000,000 decoded bytes per image. The main MCP HTTP +endpoint accepts at most 14,000,000 body bytes to accommodate base64 and JSON; +the memory endpoint remains limited to 2,000,000 bytes. Upload validation checks +size and recognized image headers against the declared type. It does not decode +or sanitize the complete image, or establish that a downstream decoder will +accept it. JPEG uploads use the canonical `image/jpeg` type. + +Identical live uploads by the same operator to the same routine/workspace return +the same reference. Uploading alone does not start an agent. Prompt submission +uses the existing start/resume/queue operation, atomically retaining the image +reference with the message before provider admission. Its idempotency fingerprint +uses the submitted text and reference, not a workspace path. Retrying an accepted +message with the same key and input returns its retained receipt even after bytes +expire or the routine moves; it does not start another turn. A changed text or +reference under that key conflicts. New submissions require available bytes in +the current routine workspace. + +Cleanup follows the configured upload retention (30 days by default). Abandoned +uploads age from upload time. Queued, executing, waiting-for-input and +waiting-for-approval messages protect their image regardless of age. After the +last referencing message settles, retention runs from its last update. Cleanup +keeps reference metadata and message receipts after removing bytes, so old +receipts remain meaningful but retrieving an expired image fails. Disabling +upload retention disables this cleanup. This is an age policy, not a storage +quota or a promise that bytes survive manual filesystem changes. diff --git a/lib/custode/janitor.ex b/lib/custode/janitor.ex index 523cbdb4..f78dd41d 100644 --- a/lib/custode/janitor.ex +++ b/lib/custode/janitor.ex @@ -24,6 +24,7 @@ defmodule Custode.Janitor do import Ecto.Query, only: [from: 2] + alias Custode.Operator.Attachments alias Custode.Repo alias Custode.Workflow.ExecutionObservation alias Custode.Workflow.Results @@ -314,29 +315,39 @@ defmodule Custode.Janitor do rows |> Enum.map(&Map.get(&1, field)) |> Enum.reject(&is_nil/1) |> Enum.sum() end - # #39/#180: content-hashed screenshots under /uploads age out - # once their turn is long past; the prompt text carries the path, so a - # very old feed entry may reference a pruned file -- retention matches - # the feed's own story, not forever. + # Managed images are retained by durable message state, not file mtime. + # Legacy files still age out, except paths retained by active legacy messages. defp prune_uploads do case retention(:uploads_days) do nil -> 0 days -> + managed = Attachments.prune(days) + protected = Attachments.protected_paths() cutoff = System.os_time(:second) - days * 86_400 - Custode.Routine.all() - |> Enum.flat_map(fn routine -> - routine.workspace |> Path.expand() |> Path.join("uploads/*") |> Path.wildcard() - end) - |> Enum.count(&stale_upload_removed?(&1, cutoff)) + legacy = + Custode.Routine.all() + |> Enum.flat_map(fn routine -> + routine.workspace + |> Path.expand() + |> Path.join("uploads/*") + |> Path.wildcard() + |> Enum.map(&{routine.workspace, &1}) + end) + |> Enum.count(fn {workspace, path} -> + not MapSet.member?(protected, path) and + Attachments.safe_file?(workspace, path) and stale_upload_removed?(path, cutoff) + end) + + managed + legacy end end defp stale_upload_removed?(path, cutoff) do - case File.stat(path, time: :posix) do - {:ok, %{mtime: mtime}} when mtime < cutoff -> File.rm(path) == :ok + case File.lstat(path, time: :posix) do + {:ok, %{type: :regular, mtime: mtime}} when mtime < cutoff -> File.rm(path) == :ok _fresh_or_gone -> false end end diff --git a/lib/custode/mcp/attachment_tools.ex b/lib/custode/mcp/attachment_tools.ex new file mode 100644 index 00000000..d23f9726 --- /dev/null +++ b/lib/custode/mcp/attachment_tools.ex @@ -0,0 +1,118 @@ +defmodule Custode.MCP.AttachmentTools do + @moduledoc false + + alias Custode.MCP.Tools + alias Custode.Operator.Attachments + + @max_base64_bytes 13_333_336 + @media_types ~w(image/png image/jpeg image/gif image/webp) + @public_fields ~w(attachment_id agent_id media_type byte_count sha256 retention created_at)a + + def upload_input do + object(%{ + "agent_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 1_024}, + "media_type" => %{"type" => "string", "enum" => @media_types}, + "data_base64" => %{"type" => "string", "minLength" => 1, "maxLength" => @max_base64_bytes} + }) + end + + def read_input do + object(%{"attachment_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 128}}) + end + + def output(mode) do + properties = %{ + "attachment_id" => %{"type" => "string"}, + "agent_id" => %{"type" => "string"}, + "media_type" => %{"type" => "string", "enum" => @media_types}, + "byte_count" => %{"type" => "integer", "minimum" => 1, "maximum" => 10_000_000}, + "sha256" => %{"type" => "string", "minLength" => 64, "maxLength" => 64}, + "retention" => %{"type" => "string"}, + "created_at" => %{"type" => "string"} + } + + properties = + if mode == :read, + do: + Map.put(properties, "data_base64", %{ + "type" => "string", + "maxLength" => @max_base64_bytes + }), + else: properties + + object(properties) + end + + def verified_operator(frame) do + case frame.assigns[:custode_identity] do + %{kind: :operator, id: id} = actor when is_binary(id) and id != "" -> {:ok, actor} + _missing_or_nonhuman -> {:error, :human_identity_required} + end + end + + def upload(params, frame) do + with {:ok, actor} <- verified_operator(frame), + {:ok, bytes} <- decode(params.data_base64), + {:ok, attachment} <- Attachments.upload(actor, params.agent_id, params.media_type, bytes) do + success(frame, attachment, :upload) + else + {:error, reason} -> refusal(frame, reason) + end + end + + def read(params, frame) do + with {:ok, actor} <- verified_operator(frame), + {:ok, attachment} <- Attachments.read(actor, params.attachment_id) do + success(frame, attachment, :read) + else + {:error, reason} -> refusal(frame, reason) + end + end + + def refusal(frame, reason), do: Tools.fail(frame, Attachments.error_message(reason)) + + defp decode(encoded) when is_binary(encoded) and byte_size(encoded) <= @max_base64_bytes do + case Base.decode64(encoded) do + {:ok, bytes} when byte_size(bytes) <= 10_000_000 -> {:ok, bytes} + {:ok, _oversize} -> {:error, :too_large} + :error -> {:error, :invalid_image} + end + end + + defp decode(_encoded), do: {:error, :too_large} + + defp success(frame, attachment, mode) do + fields = if mode == :read, do: [:data_base64 | @public_fields], else: @public_fields + result = Map.new(fields, &{Atom.to_string(&1), Map.fetch!(attachment, &1)}) + {:reply, Snodo.Result.structured(result), frame} + end + + defp object(properties) do + %{ + "type" => "object", + "additionalProperties" => false, + "required" => Map.keys(properties), + "properties" => properties + } + end +end + +defmodule Custode.MCP.AttachmentTools.Upload do + @moduledoc "Upload one bounded PNG, JPEG, GIF or WebP image for a configured routine as its verified human operator. Returns an owned attachment reference for prompt_agent; uploading does not start work. The server validates size and image header, not full codec validity." + use Custode.MCP.Tool, name: "attachment_upload", strict_arguments: true + alias Custode.MCP.AttachmentTools + input_schema(AttachmentTools.upload_input()) + output_schema(AttachmentTools.output(:upload)) + @impl true + def execute(params, frame), do: AttachmentTools.upload(params, frame) +end + +defmodule Custode.MCP.AttachmentTools.Read do + @moduledoc "Read an image by its server-owned attachment reference as the same verified human operator who uploaded it. Returns bounded metadata and Base64 bytes, never a selectable filesystem path. Reading does not start work." + use Custode.MCP.Tool, name: "attachment_read", strict_arguments: true + alias Custode.MCP.AttachmentTools + input_schema(AttachmentTools.read_input()) + output_schema(AttachmentTools.output(:read)) + @impl true + def execute(params, frame), do: AttachmentTools.read(params, frame) +end diff --git a/lib/custode/mcp/server.ex b/lib/custode/mcp/server.ex index 741dd300..a7dcf55d 100644 --- a/lib/custode/mcp/server.ex +++ b/lib/custode/mcp/server.ex @@ -5,6 +5,8 @@ defmodule Custode.MCP.Server do Custode.MCP.Tools.AgentStatus, Custode.MCP.Tools.StartAgent, Custode.MCP.Tools.PromptAgent, + Custode.MCP.AttachmentTools.Upload, + Custode.MCP.AttachmentTools.Read, Custode.MCP.Tools.AwaitAgent, Custode.MCP.Tools.AgentHistory, Custode.MCP.Tools.ApproveAction, diff --git a/lib/custode/mcp/snodo.ex b/lib/custode/mcp/snodo.ex index 21b0a87a..a4b81d70 100644 --- a/lib/custode/mcp/snodo.ex +++ b/lib/custode/mcp/snodo.ex @@ -41,6 +41,7 @@ defmodule Custode.MCP.Snodo do runtime: configured_runtime, executor: @executor, path: path, + max_body_bytes: if(path == "/mcp", do: 14_000_000, else: 2_000_000), request_timeout: @request_timeout )} end diff --git a/lib/custode/mcp/tool_policy.ex b/lib/custode/mcp/tool_policy.ex index c53f3262..9d1bdb72 100644 --- a/lib/custode/mcp/tool_policy.ex +++ b/lib/custode/mcp/tool_policy.ex @@ -48,6 +48,8 @@ defmodule Custode.MCP.ToolPolicy do "agent_history" => :read, "start_agent" => :delegate, "prompt_agent" => :delegate, + "attachment_upload" => :operator, + "attachment_read" => :operator, "approve_action" => :delegate, "reject_action" => :delegate, "run_job" => :delegate, diff --git a/lib/custode/mcp/tools.ex b/lib/custode/mcp/tools.ex index 6a60badc..5feb4a40 100644 --- a/lib/custode/mcp/tools.ex +++ b/lib/custode/mcp/tools.ex @@ -437,19 +437,27 @@ defmodule Custode.MCP.Tools.PromptAgent do this is the answer to its pending question; if it is busy, the prompt queues. Follow with await_agent to see the outcome. """ - use Custode.MCP.Tool, name: "prompt_agent" + use Custode.MCP.Tool, name: "prompt_agent", strict_arguments: true import Custode.MCP.Tools alias Custode.{AgentHandoff, Agents, OperatorMessages} + alias Custode.MCP.AttachmentTools alias Custode.Operator.Actions input_schema(%{ "properties" => %{ "agent_id" => %{"type" => "string"}, + "attachment_id" => %{ + "description" => + "one image reference owned by the verified operator and bound to this routine", + "type" => "string", + "minLength" => 1, + "maxLength" => 128 + }, "idempotency_key" => %{ "description" => "deduplicate this caller's delivery to this agent", - "type" => "string" + "type" => ["string", "null"] }, "prompt" => %{"type" => "string"} }, @@ -458,6 +466,13 @@ defmodule Custode.MCP.Tools.PromptAgent do }) @impl true + def execute(%{agent_id: agent_id, prompt: prompt, attachment_id: _id} = params, frame) do + case AttachmentTools.verified_operator(frame) do + {:ok, _actor} -> operator_prompt(agent_id, prompt, params, frame) + {:error, reason} -> AttachmentTools.refusal(frame, reason) + end + end + def execute(%{agent_id: agent_id, prompt: prompt} = params, frame) do case check_delegated_target(frame, agent_id, :manage) do :ok -> prompt(agent_id, prompt, params, frame) @@ -484,6 +499,11 @@ defmodule Custode.MCP.Tools.PromptAgent do idempotency_key: params[:idempotency_key] ] + opts = + if Map.has_key?(params, :attachment_id), + do: Keyword.put(opts, :attachment_id, params.attachment_id), + else: opts + case Actions.message_with_receipt(agent_id, prompt, opts) do {:ok, message, disposition} -> reply_message(message, disposition, frame) {:error, {:refused, message, _reason}} -> reply_message(message, :created, frame) diff --git a/lib/custode/operator/actions.ex b/lib/custode/operator/actions.ex index fcc02759..a6d7e4a2 100644 --- a/lib/custode/operator/actions.ex +++ b/lib/custode/operator/actions.ex @@ -70,20 +70,18 @@ defmodule Custode.Operator.Actions do |> submit_message(agent_id, opts) end - defp submit_message("", _agent_id, _opts), do: {:error, :empty} - defp submit_message(text, agent_id, opts) do agent_id - |> OperatorMessages.submit(text, opts, &deliver_message(agent_id, text, opts, &1)) - |> record_prompt_submission(agent_id, text) + |> OperatorMessages.submit(text, opts, &deliver_message(agent_id, &1.prompt, opts, &1)) + |> record_prompt_submission(agent_id) end - defp record_prompt_submission({:ok, _message, :created} = result, agent_id, text) do - :ok = Custode.Feed.record_prompted(agent_id, text) + defp record_prompt_submission({:ok, message, :created} = result, agent_id) do + :ok = Custode.Feed.record_prompted(agent_id, message.prompt) result end - defp record_prompt_submission(result, _agent_id, _text), do: result + defp record_prompt_submission(result, _agent_id), do: result defp deliver_message(agent_id, text, opts, message) do case Routine.get(agent_id) do diff --git a/lib/custode/operator/attachment.ex b/lib/custode/operator/attachment.ex new file mode 100644 index 00000000..d3638a86 --- /dev/null +++ b/lib/custode/operator/attachment.ex @@ -0,0 +1,32 @@ +defmodule Custode.Operator.Attachment do + @moduledoc "Server-owned operator image metadata. Expired references remain as tombstones." + + use Ecto.Schema + + import Ecto.Changeset + + schema "operator_attachments" do + field(:attachment_id, :string) + field(:owner_id, :string) + field(:agent_id, :string) + field(:workspace, :string) + field(:path, :string) + field(:sha256, :string) + field(:media_type, :string) + field(:byte_count, :integer) + field(:deleted_at, :utc_datetime_usec) + timestamps(type: :utc_datetime_usec) + end + + def changeset(attrs) do + fields = ~w(attachment_id owner_id agent_id workspace path sha256 media_type byte_count)a + + %__MODULE__{} + |> cast(attrs, fields) + |> validate_required(fields) + |> unique_constraint(:attachment_id) + |> unique_constraint([:owner_id, :agent_id, :workspace, :sha256], + name: :operator_attachments_live_content_index + ) + end +end diff --git a/lib/custode/operator/attachments.ex b/lib/custode/operator/attachments.ex index 24ae9f41..3c47f998 100644 --- a/lib/custode/operator/attachments.ex +++ b/lib/custode/operator/attachments.ex @@ -1,53 +1,141 @@ defmodule Custode.Operator.Attachments do @moduledoc """ - An image attached to a message: where it is stored and how it reaches the - agent (#180), in one place (#450). - - A dropped image reaches the agent as a PATH, not as an attachment. claude - reads images natively through Read, and a file in the routine's own - workspace is already inside its readable world, so this needs no wrapper and - no protocol change. The path is absolute because a routine's working - directory is not always its workspace: a repo-tied routine runs in the - checkout, and a relative `uploads/` would not resolve from there. - - This is the code that writes an operator-supplied file to disk, which is why - there is one copy of it. Two rules keep that write boring: - - * the stored name is a hash of the CONTENT. The same screenshot dropped - twice is one file, and a client filename never steers the write. - * the extension comes from an allowlist. Anything else is stored as `.png`. + Bounded operator image storage shared by the UI and MCP. + + Public references bind an immutable file to its uploader, routine and current + workspace. Image signatures identify the supported formats; they do not fully + decode or certify an image. Local processes with workspace access can still + replace files, so resolution checks the size and complete content digest. + + Prompt admission resolves a reference inside the same immediate transaction + that inserts its durable message. Retention checks use that transaction too: + active messages pin images, then terminal messages receive the normal retention + window. Expiration commits a tombstone before unlinking; failed unlinks are + retried on the next prune. Database and filesystem writes are not one atomic + operation. An interrupted upload may leave an unreferenced file for the janitor. """ + import Ecto.Query, only: [from: 2] + + alias Custode.Operator.Attachment + alias Custode.{OperatorMessage, Repo, Routine} + @image_types ~w(.png .jpg .jpeg .gif .webp) @max_bytes 10_000_000 + @active_statuses ~w(queued executing waiting_for_input waiting_for_approval) + @retention "Active messages retain the image. Otherwise the configured upload retention applies " <> + "from upload or latest message update (30 days by default). Expired references cannot be reused." + @extensions %{ + "image/png" => ".png", + "image/jpeg" => ".jpg", + "image/gif" => ".gif", + "image/webp" => ".webp" + } - @doc "The extensions an upload may carry. Also the `accept:` list for the form." - @spec image_types() :: [String.t()] def image_types, do: @image_types - - @doc "The largest image accepted, in bytes." - @spec max_bytes() :: pos_integer() def max_bytes, do: @max_bytes - @doc """ - Copy an uploaded temp file into `routine`'s `workspace/uploads/` and return - the absolute path it now lives at. - """ - @spec store!(map(), Path.t(), String.t()) :: Path.t() - def store!(%{workspace: workspace}, tmp_path, client_name) do - dir = Path.join(Path.expand(workspace), "uploads") - File.mkdir_p!(dir) + @doc "Store one bounded image; live identical uploads by the same owner and target reuse a reference." + def upload(actor, agent_id, media_type, bytes) do + with :ok <- human(actor), + :ok <- validate_image(media_type, bytes) do + transaction(fn -> upload_locked(actor, agent_id, media_type, bytes) end) + end + end - dest = Path.join(dir, name(tmp_path, client_name)) - File.cp!(tmp_path, dest) - dest + @doc "The UI adapter for a temporary upload; client filenames select a format, never a path." + def upload_file(actor, agent_id, tmp_path, client_name) do + with :ok <- human(actor), + {:ok, media_type} <- media_type(client_name), + {:ok, bytes} <- bounded_read(tmp_path) do + upload(actor, agent_id, media_type, bytes) + end end - @doc """ - The message as the agent receives it: the operator's text, then one line per - image telling the agent where it is and to read it first. - """ - @spec compose(String.t(), [Path.t()]) :: String.t() + @doc "Read an owned reference as base64, serialized against expiration. Never exposes a path." + def read(actor, attachment_id) do + with :ok <- human(actor) do + transaction(fn -> read_locked(actor, attachment_id) end) + end + end + + @doc "Resolve inside the caller's immediate message-admission transaction." + def resolve(actor, attachment_id, target_agent_id) do + with :ok <- human(actor), + :ok <- require_transaction(), + {:ok, row} <- owned(actor, attachment_id), + :ok <- target(row, target_agent_id), + :ok <- current_workspace(row), + {:ok, _bytes} <- contents(row) do + {:ok, row.path} + end + end + + @doc "Expire unpinned images and retry unlinking tombstones. Returns newly expired references." + def prune(days) when is_integer(days) and days >= 0 do + cutoff = DateTime.add(DateTime.utc_now(), -days, :day) + + case transaction(fn -> {:ok, retire_locked(cutoff)} end) do + {:ok, count} -> + Repo.all(from(a in Attachment, where: not is_nil(a.deleted_at))) + |> Enum.each(&unlink_retired/1) + + count + + {:error, _reason} -> + 0 + end + end + + @doc "Paths the legacy mtime janitor must not remove, including managed tombstones." + def protected_paths do + managed = Repo.all(from(a in Attachment, select: a.path)) + + legacy = + Repo.all( + from(m in OperatorMessage, + where: is_nil(m.attachment_id) and m.status in ^@active_statuses, + select: m.prompt + ) + ) + |> Enum.flat_map(&prompt_paths/1) + + MapSet.new(managed ++ legacy) + end + + @doc "A regular file directly inside a real workspace/uploads directory, with no final symlinks." + def safe_file?(workspace, path) when is_binary(workspace) and is_binary(path) do + with {:ok, dir} <- upload_dir(workspace, :existing), + true <- Path.dirname(path) == dir, + {:ok, %{type: :regular}} <- File.lstat(path) do + true + else + _unsafe -> false + end + end + + def safe_file?(_workspace, _path), do: false + + @doc "The dashboard must not serve a managed file whose expiration already committed." + def retired_path?(path) do + Repo.exists?(from(a in Attachment, where: a.path == ^path and not is_nil(a.deleted_at))) + end + + @doc "Strict legacy path adapter. New message surfaces use upload_file/4 and durable references." + def store!(%{workspace: workspace}, tmp_path, client_name) do + with {:ok, media_type} <- media_type(client_name), + {:ok, bytes} <- bounded_read(tmp_path), + :ok <- validate_image(media_type, bytes), + {:ok, dir} <- upload_dir(workspace, :create), + path = Path.join(dir, digest(bytes) <> Map.fetch!(@extensions, media_type)), + :ok <- store_legacy(path, bytes) do + path + else + {:error, reason} -> raise ArgumentError, error_message(reason) + end + end + + @doc "The provider receives the text followed by its server-resolved image path." def compose(text, []), do: text def compose(text, paths) do @@ -56,18 +144,320 @@ defmodule Custode.Operator.Attachments do |> String.trim() end - defp name(tmp_path, client_name) do - hash = - :sha256 - |> :crypto.hash(File.read!(tmp_path)) - |> Base.encode16(case: :lower) - |> binary_part(0, 16) + def error_message(:human_identity_required), + do: "An authenticated operator identity is required." + + def error_message(:no_such_routine), do: "The target agent is not configured." + def error_message(:unsupported_media_type), do: "Choose a PNG, JPEG, GIF or WebP image." + def error_message(:invalid_image), do: "The image header does not match its declared format." + def error_message(:too_large), do: "The image exceeds the 10,000,000-byte limit." + def error_message(:invalid_attachment), do: "Provide a valid attachment reference." + def error_message(:attachment_not_found), do: "The attachment reference was not found." + def error_message(:attachment_forbidden), do: "This attachment belongs to another operator." + def error_message(:attachment_target_mismatch), do: "This attachment belongs to another agent." + + def error_message(:attachment_workspace_changed), + do: "The agent's workspace changed. Upload the image again." + + def error_message(:attachment_expired), do: "This attachment expired. Upload the image again." + + def error_message(:attachment_unavailable), + do: "The stored image is missing or changed. Upload it again." + + def error_message(:unsafe_path), + do: "The image storage path is not a regular file or directory." + + def error_message(:transaction_required), + do: "Attachment resolution requires message admission." + + def error_message(_reason), do: "The image could not be stored or retrieved. Try again." + + defp upload_locked(actor, agent_id, media_type, bytes) do + with {:ok, routine} <- routine(agent_id), + {:ok, dir} <- upload_dir(routine.workspace, :create) do + store_managed(actor.id, routine.id, dir, media_type, bytes) + end + end + + defp read_locked(actor, attachment_id) do + with {:ok, row} <- owned(actor, attachment_id), + :ok <- current_workspace(row), + {:ok, bytes} <- contents(row) do + {:ok, Map.put(public(row), :data_base64, Base.encode64(bytes))} + end + end + + defp human(%{kind: :operator, id: id}) when is_binary(id) and byte_size(id) > 0, do: :ok + defp human(_actor), do: {:error, :human_identity_required} + + defp require_transaction do + if Repo.in_transaction?(), do: :ok, else: {:error, :transaction_required} + end + + defp routine(agent_id) when is_binary(agent_id) do + case Routine.get(agent_id) do + nil -> {:error, :no_such_routine} + value -> {:ok, value} + end + end + + defp routine(_agent_id), do: {:error, :no_such_routine} + + defp transaction(fun) do + if Repo.in_transaction?() do + {:error, :storage_failed} + else + Repo.transaction(fn -> unwrap(fun.()) end, mode: :immediate) + end + end + + defp unwrap({:ok, value}), do: value + defp unwrap({:error, reason}), do: Repo.rollback(reason) + + defp owned(actor, id) when is_binary(id) and byte_size(id) > 0 and byte_size(id) <= 128 do + case Repo.get_by(Attachment, attachment_id: id) do + nil -> {:error, :attachment_not_found} + %{owner_id: owner} when owner != actor.id -> {:error, :attachment_forbidden} + %{deleted_at: deleted} when not is_nil(deleted) -> {:error, :attachment_expired} + row -> {:ok, row} + end + end + + defp owned(_actor, _id), do: {:error, :invalid_attachment} + defp target(%{agent_id: target}, target), do: :ok + defp target(_row, _target), do: {:error, :attachment_target_mismatch} + + defp current_workspace(row) do + with {:ok, routine} <- routine(row.agent_id) do + if Path.expand(routine.workspace) == row.workspace, + do: :ok, + else: {:error, :attachment_workspace_changed} + end + end + + defp store_managed(owner, agent_id, dir, media_type, bytes) do + attrs = %{ + owner_id: owner, + agent_id: agent_id, + workspace: Path.dirname(dir), + sha256: digest(bytes) + } + + existing = + Repo.one( + from(a in Attachment, + where: + a.owner_id == ^owner and a.agent_id == ^agent_id and + a.workspace == ^attrs.workspace and a.sha256 == ^attrs.sha256 and + is_nil(a.deleted_at) + ) + ) + + case existing do + nil -> insert_image(attrs, dir, media_type, bytes) + row -> reuse_or_replace(row, attrs, dir, media_type, bytes) + end + end + + defp reuse_or_replace(row, attrs, dir, media_type, bytes) do + case contents(row) do + {:ok, _bytes} -> + {:ok, public(row)} + + {:error, :attachment_unavailable} -> + row |> Ecto.Changeset.change(deleted_at: DateTime.utc_now()) |> Repo.update!() + insert_image(attrs, dir, media_type, bytes) + end + end + + defp insert_image(attrs, dir, media_type, bytes) do + id = Ecto.UUID.generate() + path = Path.join(dir, "attachment-" <> id <> Map.fetch!(@extensions, media_type)) + + with :ok <- exclusive_write(path, bytes) do + attrs = + Map.merge(attrs, %{ + attachment_id: id, + path: path, + media_type: media_type, + byte_count: byte_size(bytes) + }) + + case attrs |> Attachment.changeset() |> Repo.insert() do + {:ok, row} -> + {:ok, public(row)} + + {:error, _changeset} -> + File.rm(path) + {:error, :storage_failed} + end + end + end + + defp public(row) do + %{ + attachment_id: row.attachment_id, + agent_id: row.agent_id, + media_type: row.media_type, + byte_count: row.byte_count, + sha256: row.sha256, + retention: @retention, + created_at: DateTime.to_iso8601(row.inserted_at) + } + end + + defp contents(row) do + if safe_file?(row.workspace, row.path) do + with {:ok, bytes} <- bounded_read(row.path), + true <- byte_size(bytes) == row.byte_count and digest(bytes) == row.sha256 do + {:ok, bytes} + else + _unavailable -> {:error, :attachment_unavailable} + end + else + {:error, :attachment_unavailable} + end + end + + defp validate_image(media_type, bytes) when is_binary(bytes) do + cond do + not Map.has_key?(@extensions, media_type) -> {:error, :unsupported_media_type} + byte_size(bytes) > @max_bytes -> {:error, :too_large} + signature(bytes) != media_type -> {:error, :invalid_image} + true -> :ok + end + end + + defp validate_image(_media_type, _bytes), do: {:error, :invalid_image} + defp signature(<<137, "PNG\r\n", 26, "\n", _rest::binary>>), do: "image/png" + defp signature(<<255, 216, 255, _rest::binary>>), do: "image/jpeg" + defp signature(<<"GIF87a", _rest::binary>>), do: "image/gif" + defp signature(<<"GIF89a", _rest::binary>>), do: "image/gif" + defp signature(<<"RIFF", _size::binary-size(4), "WEBP", _rest::binary>>), do: "image/webp" + defp signature(_bytes), do: nil + + defp media_type(name) when is_binary(name) do + case name |> Path.extname() |> String.downcase() do + ".png" -> {:ok, "image/png"} + ext when ext in [".jpg", ".jpeg"] -> {:ok, "image/jpeg"} + ".gif" -> {:ok, "image/gif"} + ".webp" -> {:ok, "image/webp"} + _extension -> {:error, :unsupported_media_type} + end + end + + defp media_type(_name), do: {:error, :unsupported_media_type} - hash <> extension(client_name) + defp bounded_read(path) when is_binary(path) do + with {:ok, stat} <- File.lstat(path), + :ok <- readable(stat), + {:ok, result} <- File.open(path, [:read, :binary], &IO.binread(&1, @max_bytes + 1)) do + bounded_result(result) + else + {:error, reason} when reason in [:too_large, :unsafe_path] -> {:error, reason} + _unavailable -> {:error, :attachment_unavailable} + end end - defp extension(client_name) do - extension = client_name |> Path.extname() |> String.downcase() - if extension in @image_types, do: extension, else: ".png" + defp bounded_read(_path), do: {:error, :unsafe_path} + defp readable(%{type: :regular, size: size}) when size > @max_bytes, do: {:error, :too_large} + defp readable(%{type: :regular}), do: :ok + defp readable(_stat), do: {:error, :unsafe_path} + + defp bounded_result(bytes) when is_binary(bytes) and byte_size(bytes) > @max_bytes, + do: {:error, :too_large} + + defp bounded_result(bytes) when is_binary(bytes), do: {:ok, bytes} + defp bounded_result(:eof), do: {:ok, ""} + defp bounded_result(_other), do: {:error, :attachment_unavailable} + + defp upload_dir(workspace, mode) do + workspace = Path.expand(workspace) + dir = Path.join(workspace, "uploads") + + with :ok <- ensure_workspace(workspace, mode), + :ok <- real_directory(workspace), + :ok <- ensure_uploads(dir, mode), + :ok <- real_directory(dir) do + {:ok, dir} + end + end + + defp ensure_workspace(path, :create), do: file_result(File.mkdir_p(path)) + defp ensure_workspace(_path, :existing), do: :ok + defp ensure_uploads(path, :create), do: file_result(File.mkdir(path)) + defp ensure_uploads(_path, :existing), do: :ok + defp file_result(:ok), do: :ok + defp file_result({:error, :eexist}), do: :ok + defp file_result(_error), do: {:error, :unsafe_path} + + defp real_directory(path) do + case File.lstat(path) do + {:ok, %{type: :directory}} -> :ok + _unsafe -> {:error, :unsafe_path} + end + end + + defp exclusive_write(path, bytes) do + case File.write(path, bytes, [:binary, :exclusive]) do + :ok -> :ok + {:error, _reason} -> {:error, :storage_failed} + end + end + + defp store_legacy(path, bytes) do + case File.lstat(path) do + {:error, :enoent} -> + exclusive_write(path, bytes) + + {:ok, %{type: :regular}} -> + if bounded_read(path) == {:ok, bytes}, do: :ok, else: {:error, :attachment_unavailable} + + _unsafe -> + {:error, :unsafe_path} + end + end + + defp digest(bytes), do: :crypto.hash(:sha256, bytes) |> Base.encode16(case: :lower) + + defp retire_locked(cutoff) do + candidates = + Repo.all(from(a in Attachment, where: is_nil(a.deleted_at) and a.inserted_at < ^cutoff)) + + now = DateTime.utc_now() + + Enum.count(candidates, fn row -> + if unpinned?(row.attachment_id, cutoff) do + row |> Ecto.Changeset.change(deleted_at: now) |> Repo.update!() + true + else + false + end + end) + end + + defp unpinned?(id, cutoff) do + not Repo.exists?( + from(m in OperatorMessage, + where: + m.attachment_id == ^id and (m.status in ^@active_statuses or m.updated_at >= ^cutoff) + ) + ) + end + + defp unlink_retired(row) do + if safe_file?(row.workspace, row.path), do: File.rm(row.path) + end + + defp prompt_paths(prompt) do + prompt + |> String.split("\n") + |> Enum.flat_map(fn + "attached image: " <> rest -> + [String.replace_suffix(rest, " -- Read it before answering", "")] + + _line -> + [] + end) end end diff --git a/lib/custode/operator_message.ex b/lib/custode/operator_message.ex index e4a64161..4016f729 100644 --- a/lib/custode/operator_message.ex +++ b/lib/custode/operator_message.ex @@ -16,6 +16,7 @@ defmodule Custode.OperatorMessage do field(:idempotency_key, :string) field(:prompt_hash, :string) field(:prompt, :string) + field(:attachment_id, :string) field(:provider_correlation_id, :string) field(:continues_message_id, :string) field(:status, :string, default: "queued") @@ -79,6 +80,7 @@ defmodule Custode.OperatorMessage do :idempotency_key, :prompt_hash, :prompt, + :attachment_id, :provider_correlation_id, :continues_message_id, :status, diff --git a/lib/custode/operator_messages.ex b/lib/custode/operator_messages.ex index 5b247ab2..c4aa21fa 100644 --- a/lib/custode/operator_messages.ex +++ b/lib/custode/operator_messages.ex @@ -12,6 +12,7 @@ defmodule Custode.OperatorMessages do import Ecto.Query, only: [from: 2] alias Custode.{AgentHandoff, Agents, OperatorMessage, Repo} + alias Custode.Operator.Attachments @handler_id "custode-operator-messages" @events [ @@ -70,11 +71,12 @@ defmodule Custode.OperatorMessages do @spec submit(String.t(), String.t(), keyword(), (OperatorMessage.t() -> term())) :: {:ok, OperatorMessage.t(), :created | :duplicate} | {:error, term()} def submit(target_agent_id, prompt, opts, deliver) when is_function(deliver, 1) do - with {:ok, caller} <- caller(opts), - {:ok, prompt} <- prompt(prompt) do + with :ok <- submission_boundary(opts[:attachment_id]), + {:ok, caller} <- caller(opts), + {:ok, prompt} <- prompt(prompt, opts[:attachment_id]) do message_id = message_id() idempotency_key = Keyword.get(opts, :idempotency_key) || message_id - prompt_hash = hash(prompt) + prompt_hash = request_hash(prompt, opts[:attachment_id]) continued = waiting_message(target_agent_id) provider_correlation_id = correlation_of(continued) || message_id @@ -87,6 +89,7 @@ defmodule Custode.OperatorMessages do idempotency_key: idempotency_key, prompt_hash: prompt_hash, prompt: prompt, + attachment_id: opts[:attachment_id], provider_correlation_id: provider_correlation_id, continues_message_id: continued && continued.message_id, provider: target_agent_id |> configured_provider() |> to_string(), @@ -222,6 +225,7 @@ defmodule Custode.OperatorMessages do %{ message_id: message.message_id, target_agent_id: message.target_agent_id, + attachment_id: message.attachment_id, caller: %{ kind: message.caller_kind, id: message.caller_id, @@ -251,6 +255,7 @@ defmodule Custode.OperatorMessages do %{ message_id: message.message_id, agent_id: message.target_agent_id, + attachment_id: message.attachment_id, status: message.status, delivery: message.delivery, provider: message.provider, @@ -583,25 +588,52 @@ defmodule Custode.OperatorMessages do :ok end + # Resolve and pin an attachment in the same writer transaction as the + # message. Cleanup uses this boundary too. Provider admission stays outside + # the transaction, so no external effect can precede durable acceptance. defp insert_or_get(attrs, prompt_hash, deliver) do - changeset = OperatorMessage.create_changeset(attrs) + case Repo.transaction(fn -> insert_or_retry(attrs, prompt_hash) end, mode: :immediate) do + {:ok, {:created, message}} -> deliver_new(message, deliver) + {:ok, {:duplicate, message}} -> {:ok, message, :duplicate} + {:ok, {:error, reason}} -> {:error, reason} + {:error, reason} -> {:error, reason} + end + end - case Repo.insert(changeset, - on_conflict: :nothing, - conflict_target: [:caller_kind, :caller_id, :target_agent_id, :idempotency_key], - returning: true - ) do - {:ok, %OperatorMessage{id: nil}} -> - duplicate(attrs, prompt_hash) + defp insert_or_retry(attrs, prompt_hash) do + attachment_id = attrs.attachment_id + + case existing_submission(attrs) do + nil -> + with {:ok, attrs} <- resolve_attachment(attrs), + {:ok, message} <- Repo.insert(OperatorMessage.create_changeset(attrs)) do + {:created, message} + else + {:error, reason} -> {:error, reason} + end - {:ok, %OperatorMessage{} = message} -> - deliver_new(message, deliver) + %{prompt_hash: ^prompt_hash, attachment_id: ^attachment_id} = message -> + # A receipt survives expired bytes and routine moves. An exact replay + # must not resolve the attachment again or invoke the provider again. + {:duplicate, message} - {:error, changeset} -> - {:error, changeset} + _different_request -> + {:error, :idempotency_conflict} end end + defp resolve_attachment(%{attachment_id: nil} = attrs), do: {:ok, attrs} + + defp resolve_attachment(%{caller_kind: "operator"} = attrs) do + actor = %{kind: :operator, id: attrs.caller_id} + + with {:ok, path} <- Attachments.resolve(actor, attrs.attachment_id, attrs.target_agent_id) do + {:ok, %{attrs | prompt: Attachments.compose(attrs.prompt, [path])}} + end + end + + defp resolve_attachment(_attrs), do: {:error, :human_identity_required} + defp scope_target(query, nil), do: query defp scope_target(query, agent_id), @@ -708,6 +740,7 @@ defmodule Custode.OperatorMessages do %{ id: message.message_id, text: message.prompt, + attachment_id: message.attachment_id, caller_id: message.caller_id, continued: not is_nil(message.continues_message_id), detail: message.detail, @@ -746,21 +779,16 @@ defmodule Custode.OperatorMessages do defp conversation_error(error) when is_map(error), do: Jason.encode!(error, pretty: true) defp conversation_error(_error), do: nil - defp duplicate(attrs, prompt_hash) do - existing = - Repo.one!( - from(m in OperatorMessage, - where: - m.caller_kind == ^attrs.caller_kind and m.caller_id == ^attrs.caller_id and - m.target_agent_id == ^attrs.target_agent_id and - m.idempotency_key == ^attrs.idempotency_key, - limit: 1 - ) + defp existing_submission(attrs) do + Repo.one( + from(m in OperatorMessage, + where: + m.caller_kind == ^attrs.caller_kind and m.caller_id == ^attrs.caller_id and + m.target_agent_id == ^attrs.target_agent_id and + m.idempotency_key == ^attrs.idempotency_key, + limit: 1 ) - - if existing.prompt_hash == prompt_hash, - do: {:ok, existing, :duplicate}, - else: {:error, :idempotency_conflict} + ) end defp deliver_new(message, deliver) do @@ -1452,13 +1480,33 @@ defmodule Custode.OperatorMessages do end end - defp prompt(text) do - case text |> to_string() |> String.trim() do - "" -> {:error, :empty} - prompt -> {:ok, prompt} + defp submission_boundary(nil), do: :ok + + defp submission_boundary(_attachment_id) do + if Repo.in_transaction?(), do: {:error, :outer_transaction_unsupported}, else: :ok + end + + defp prompt(text, attachment_id) do + text = text |> to_string() |> String.trim() + + cond do + not is_nil(attachment_id) and + (not is_binary(attachment_id) or byte_size(attachment_id) not in 1..128) -> + {:error, :invalid_attachment} + + text == "" and is_nil(attachment_id) -> + {:error, :empty} + + true -> + {:ok, text} end end + # Keep the original text-only fingerprint for existing receipt retries. + # Attachment identity is independent of its resolved provider path. + defp request_hash(prompt, nil), do: hash(prompt) + defp request_hash(prompt, id), do: hash("attachment-v1:" <> Jason.encode!([prompt, id])) + defp correlation(meta), do: meta[:correlation_id] defp job_correlation(meta), do: meta |> job_meta() |> Map.get("correlation_id") diff --git a/lib/custode/uploads.ex b/lib/custode/uploads.ex index aa542921..f0432781 100644 --- a/lib/custode/uploads.ex +++ b/lib/custode/uploads.ex @@ -3,7 +3,7 @@ defmodule Custode.Uploads do The `uploads/` directory of a routine's workspace, addressed safely. An image dropped on a prompt box (#180) is written into the routine's own - workspace as `uploads/.`, and the prompt text keeps the + workspace under `uploads/` with a server-owned filename, and the prompt text keeps the absolute path so the agent can `Read` it. This module is the one place that turns an `{agent, filename}` pair back into a file on disk and into the URL the dashboard serves it at, so the web route and the thumbnail component @@ -15,10 +15,10 @@ defmodule Custode.Uploads do the upload boxes accept, and the file must actually be a regular file. """ - @image_types ~w(.png .jpg .jpeg .gif .webp) + alias Custode.Operator.Attachments @doc "The image extensions a dropped file may carry." - def image_types, do: @image_types + def image_types, do: Attachments.image_types() @doc """ The on-disk path of one uploaded image, or `:error` when it is not @@ -27,10 +27,11 @@ defmodule Custode.Uploads do def path(agent, file) when is_binary(agent) and is_binary(file) do name = Path.basename(file) - with true <- Path.extname(String.downcase(name)) in @image_types, + with true <- Path.extname(String.downcase(name)) in Attachments.image_types(), routine when not is_nil(routine) <- Custode.Routine.get(agent), resolved = Path.join([Path.expand(routine.workspace), "uploads", name]), - true <- File.regular?(resolved) do + true <- Attachments.safe_file?(routine.workspace, resolved), + false <- Attachments.retired_path?(resolved) do {:ok, resolved} else _unreachable -> :error diff --git a/lib/custode_web/console_live.ex b/lib/custode_web/console_live.ex index 75b2cd2a..fb99ad7a 100644 --- a/lib/custode_web/console_live.ex +++ b/lib/custode_web/console_live.ex @@ -296,14 +296,13 @@ defmodule CustodeWeb.ConsoleLive do do: {:noreply, cancel_upload(socket, :image, ref)} def handle_event("message", %{"text" => text}, socket) do - # An image alone is a message. The stored paths join the text before it is - # delivered, so every delivery path (queue, resume, start) carries them. - text = Attachments.compose(text, save_images(socket)) - - case Actions.message(socket.assigns.selected, text, @opts) do - {:ok, how} -> - socket = push_event(socket, "draft:clear", %{subject: socket.assigns.selected}) - after_action(:ok, socket, sent_notice(how)) + with {:ok, attachment_opts} <- save_images(socket), + {:ok, how} <- Actions.message(socket.assigns.selected, text, @opts ++ attachment_opts) do + socket = push_event(socket, "draft:clear", %{subject: socket.assigns.selected}) + after_action(:ok, socket, sent_notice(how)) + else + {:error, {:attachment, reason}} -> + {:noreply, assign(socket, notice: Attachments.error_message(reason))} {:error, reason} -> after_action({:error, reason}, socket, nil) @@ -598,12 +597,36 @@ defmodule CustodeWeb.ConsoleLive do # A subject with no routine has no workspace to put an image in. defp save_images(%{assigns: %{subject: %{routine: %{} = routine}}} = socket) do - consume_uploaded_entries(socket, :image, fn %{path: path}, entry -> - {:ok, Attachments.store!(routine, path, entry.client_name)} - end) + case uploaded_entries(socket, :image) do + {_completed, []} -> + results = + consume_uploaded_entries(socket, :image, fn %{path: path}, entry -> + {:ok, + Attachments.upload_file( + %{kind: :operator, id: "operator"}, + routine.id, + path, + entry.client_name + )} + end) + + case results do + [] -> {:ok, []} + [{:ok, attachment}] -> {:ok, [attachment_id: attachment.attachment_id]} + [{:error, reason}] -> {:error, {:attachment, reason}} + end + + _upload_in_progress -> + {:error, {:attachment, :invalid_attachment}} + end end - defp save_images(_socket), do: [] + defp save_images(socket) do + case uploaded_entries(socket, :image) do + {[], []} -> {:ok, []} + _images -> {:error, {:attachment, :no_such_routine}} + end + end defp sent_notice(:delivered), do: "sent" defp sent_notice(:queued), do: "queued for the next safe turn" diff --git a/lib/custode_web/conversation_live.ex b/lib/custode_web/conversation_live.ex index 14a1563a..04593e76 100644 --- a/lib/custode_web/conversation_live.ex +++ b/lib/custode_web/conversation_live.ex @@ -144,12 +144,13 @@ defmodule CustodeWeb.ConversationLive do do: {:noreply, assign(socket, notice: "set up a caretaker before sending")} def handle_event("message", %{"text" => text}, socket) do - text = Attachments.compose(text, save_images(socket)) - - case Actions.message(socket.assigns.agent_id, text, @opts) do - {:ok, how} -> - socket = push_event(socket, "draft:clear", %{subject: socket.assigns.agent_id}) - after_action(:ok, socket, sent_notice(how)) + with {:ok, attachment_opts} <- save_images(socket), + {:ok, how} <- Actions.message(socket.assigns.agent_id, text, @opts ++ attachment_opts) do + socket = push_event(socket, "draft:clear", %{subject: socket.assigns.agent_id}) + after_action(:ok, socket, sent_notice(how)) + else + {:error, {:attachment, reason}} -> + {:noreply, assign(socket, notice: Attachments.error_message(reason))} {:error, reason} -> after_action({:error, reason}, socket, nil) @@ -623,12 +624,36 @@ defmodule CustodeWeb.ConversationLive do end defp save_images(%{assigns: %{subject: %{routine: %{} = routine}}} = socket) do - consume_uploaded_entries(socket, :image, fn %{path: path}, entry -> - {:ok, Attachments.store!(routine, path, entry.client_name)} - end) + case uploaded_entries(socket, :image) do + {_completed, []} -> + results = + consume_uploaded_entries(socket, :image, fn %{path: path}, entry -> + {:ok, + Attachments.upload_file( + %{kind: :operator, id: "operator"}, + routine.id, + path, + entry.client_name + )} + end) + + case results do + [] -> {:ok, []} + [{:ok, attachment}] -> {:ok, [attachment_id: attachment.attachment_id]} + [{:error, reason}] -> {:error, {:attachment, reason}} + end + + _upload_in_progress -> + {:error, {:attachment, :invalid_attachment}} + end end - defp save_images(_socket), do: [] + defp save_images(socket) do + case uploaded_entries(socket, :image) do + {[], []} -> {:ok, []} + _images -> {:error, {:attachment, :no_such_routine}} + end + end defp sent_notice(:delivered), do: "sent" defp sent_notice(:queued), do: "queued for the next safe turn" diff --git a/priv/repo/migrations/20261012151227_add_operator_attachments.exs b/priv/repo/migrations/20261012151227_add_operator_attachments.exs new file mode 100644 index 00000000..cd59e32c --- /dev/null +++ b/priv/repo/migrations/20261012151227_add_operator_attachments.exs @@ -0,0 +1,33 @@ +defmodule Custode.Repo.Migrations.AddOperatorAttachments do + use Ecto.Migration + + def change do + create table(:operator_attachments) do + add(:attachment_id, :string, null: false) + add(:owner_id, :string, null: false) + add(:agent_id, :string, null: false) + add(:workspace, :string, null: false) + add(:path, :string, null: false) + add(:sha256, :string, null: false) + add(:media_type, :string, null: false) + add(:byte_count, :integer, null: false) + add(:deleted_at, :utc_datetime_usec) + timestamps(type: :utc_datetime_usec) + end + + create(unique_index(:operator_attachments, [:attachment_id])) + + create( + unique_index(:operator_attachments, [:owner_id, :agent_id, :workspace, :sha256], + name: :operator_attachments_live_content_index, + where: "deleted_at IS NULL" + ) + ) + + alter table(:operator_messages) do + add(:attachment_id, :string) + end + + create(index(:operator_messages, [:attachment_id])) + end +end diff --git a/test/custode/attachments_test.exs b/test/custode/attachments_test.exs index 317768e2..c71b2abb 100644 --- a/test/custode/attachments_test.exs +++ b/test/custode/attachments_test.exs @@ -5,10 +5,14 @@ defmodule Custode.Operator.AttachmentsTest do alias Custode.Operator.Attachments + @png Base.decode64!( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" + ) + setup do workspace = tmp_workspace!() tmp = Path.join(System.tmp_dir!(), uid("upload")) - File.write!(tmp, "not really a png") + File.write!(tmp, @png) on_exit(fn -> File.rm(tmp) end) %{routine: %{workspace: workspace}, tmp: tmp, workspace: workspace} end @@ -19,20 +23,31 @@ defmodule Custode.Operator.AttachmentsTest do assert Path.type(path) == :absolute assert Path.dirname(path) == Path.join(Path.expand(workspace), "uploads") - assert File.read!(path) == "not really a png" + assert File.read!(path) == @png end # this is the code that writes an operator-supplied file to disk test "the client's filename never steers the write", %{routine: routine, tmp: tmp} do path = Attachments.store!(routine, tmp, "../../../etc/passwd.png") - assert Path.basename(path) =~ ~r/^[0-9a-f]{16}\.png$/ + assert Path.basename(path) =~ ~r/^[0-9a-f]{64}\.png$/ refute path =~ ".." end - test "an extension off the allowlist is stored as .png", %{routine: routine, tmp: tmp} do - assert Path.extname(Attachments.store!(routine, tmp, "payload.exe")) == ".png" - assert Path.extname(Attachments.store!(routine, tmp, "PHOTO.JPG")) == ".jpg" + test "unknown extensions and declared formats that mismatch bytes are refused", %{ + routine: routine, + tmp: tmp + } do + assert_raise ArgumentError, fn -> Attachments.store!(routine, tmp, "payload.exe") end + assert_raise ArgumentError, fn -> Attachments.store!(routine, tmp, "PHOTO.JPG") end + assert Path.extname(Attachments.store!(routine, tmp, "PHOTO.PNG")) == ".png" + end + + test "a symlinked uploads directory cannot steer storage", %{routine: routine, tmp: tmp} do + outside = tmp_workspace!() + File.ln_s!(outside, Path.join(routine.workspace, "uploads")) + assert_raise ArgumentError, fn -> Attachments.store!(routine, tmp, "a.png") end + assert File.ls!(outside) == ["inbox"] end test "the same content dropped twice is one file", %{routine: routine, tmp: tmp} do diff --git a/test/custode/janitor_test.exs b/test/custode/janitor_test.exs index f4fc11c8..bad588b3 100644 --- a/test/custode/janitor_test.exs +++ b/test/custode/janitor_test.exs @@ -4,6 +4,7 @@ defmodule Custode.JanitorTest do import Custode.TestHelpers alias Custode.Janitor + alias Custode.Operator.{Attachment, Attachments} defp perform!, do: Janitor.perform(%Oban.Job{args: %{}}) @@ -129,6 +130,47 @@ defmodule Custode.JanitorTest do assert File.exists?(fresh) end + test "old managed and legacy images stay while durable messages still need them" do + routine = routine_fixture!(tmp_workspace!()) + actor = %{kind: :operator, id: uid("janitor-image-owner")} + png = <<137, "PNG\r\n", 26, "\n", "fixture">> + + {:ok, image} = Attachments.upload(actor, routine.id, "image/png", png) + + attachment = + Custode.Repo.get_by!(Attachment, attachment_id: image.attachment_id) + + {:ok, message, :created} = + Custode.OperatorMessages.submit( + routine.id, + "inspect", + [actor: actor, attachment_id: image.attachment_id, idempotency_key: uid("janitor-image")], + fn _message -> {:ok, :queued} end + ) + + legacy = Path.join(Path.dirname(attachment.path), "legacy image.png") + File.write!(legacy, png) + prompt = Attachments.compose("inspect", [legacy]) + + {:ok, legacy_message, :created} = + Custode.OperatorMessages.submit( + routine.id, + prompt, + [actor: actor, idempotency_key: uid("legacy-image")], + fn _message -> {:ok, :queued} end + ) + + backdate!("operator_attachments", attachment.id, "inserted_at", 90) + backdate!("operator_messages", message.id, "updated_at", 90) + backdate!("operator_messages", legacy_message.id, "updated_at", 90) + File.touch!(attachment.path, System.os_time(:second) - 90 * 86_400) + File.touch!(legacy, System.os_time(:second) - 90 * 86_400) + :ok = perform!() + assert File.exists?(attachment.path) + assert File.exists?(legacy) + assert Custode.Repo.get!(Attachment, attachment.id).deleted_at == nil + end + test "compacted journal entries retire only once aged out; live entries are immortal (#214)" do routine = routine_fixture!(tmp_workspace!()) diff --git a/test/custode/mcp_attachment_test.exs b/test/custode/mcp_attachment_test.exs new file mode 100644 index 00000000..c42070b9 --- /dev/null +++ b/test/custode/mcp_attachment_test.exs @@ -0,0 +1,425 @@ +defmodule Custode.MCPAttachmentTest do + use ExUnit.Case, async: false + + import Custode.TestHelpers, only: [put_env!: 2, tmp_workspace!: 0, uid: 1] + import Ecto.Query, only: [from: 2] + + alias Custode.MCP.AttachmentTools + alias Custode.MCP.CallContext + alias Custode.MCP.Identity + alias Custode.MCP.Snodo, as: SnodoTransport + alias Custode.MCP.ToolPolicy + alias Custode.MCP.Tools + alias Custode.Operator.Attachments + alias Custode.OperatorMessage + alias Custode.Repo + + @png Base.decode64!( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" + ) + @tools ~w(attachment_upload attachment_read) + + setup do + worker = uid("attachment-worker") + other = uid("attachment-other") + caretaker = uid("attachment-caretaker") + ids = [worker, other, caretaker] + + put_env!( + :routines, + for id <- ids do + %{ + id: id, + role: if(id == caretaker, do: :caretaker, else: :backlog_worker), + provider: :claude, + cron: :manual, + workspace: tmp_workspace!(), + prompt: "fixture only", + on_note: :ignore + } + end + ) + + on_exit(fn -> + for id <- ids, do: ObanClaude.Agent.stop_agent(id) + + Repo.delete_all( + from(j in Oban.Job, + where: fragment("json_extract(?, '$.agent_id')", j.meta) in ^ids + ) + ) + + Repo.delete_all(from(m in OperatorMessage, where: m.target_agent_id in ^ids)) + for id <- ids, do: Repo.query!("DELETE FROM operator_attachments WHERE agent_id = ?", [id]) + end) + + assert Application.get_env(:custode, :oban_queues) == [] + assert Application.get_env(:custode, :scheduler_autostart) == false + %{worker: worker, other: other, caretaker: caretaker} + end + + test "direct handlers and attachment prompts require a verified operator", ctx do + for actor <- [ + nil, + %{}, + %{kind: :operator, id: ""}, + %{kind: :routine, id: ctx.caretaker}, + %{kind: :sub_agent, id: uid("helper")} + ] do + frame = %CallContext{assigns: %{custode_identity: actor}} + + calls = [ + {AttachmentTools.Upload, + %{agent_id: ctx.worker, media_type: "image/png", data_base64: Base.encode64(@png)}}, + {AttachmentTools.Read, %{attachment_id: "unknown"}}, + {Tools.PromptAgent, + %{agent_id: ctx.worker, prompt: "do not drop image", attachment_id: "unknown"}} + ] + + for {module, params} <- calls do + assert {:reply, %Snodo.Result{kind: :error, value: message}, ^frame} = + module.execute(params, frame) + + assert message == Attachments.error_message(:human_identity_required) + end + end + + assert Repo.all(from(m in OperatorMessage, where: m.target_agent_id == ^ctx.worker)) == [] + assert :offline = Custode.Agents.live_provider(ctx.worker) + for name <- @tools, do: assert(ToolPolicy.fetch(name) == {:ok, :operator}) + end + + for protocol <- ["2025-11-25", "2026-07-28"] do + @protocol protocol + + test "owned attachment upload, read and prompt correlation over HTTP #{@protocol}", ctx do + actor = %{kind: :operator, id: uid("attachment-human")} + token = Identity.mint(actor.kind, actor.id) + initialize(token, @protocol) + discovered = discover(token, @protocol) + + for {name, module} <- [ + {"attachment_upload", AttachmentTools.Upload}, + {"attachment_read", AttachmentTools.Read} + ] do + tool = Enum.find(discovered, &(&1["name"] == name)) + assert tool["inputSchema"] == module.input_schema() + assert tool["outputSchema"]["properties"] == module.output_schema()["properties"] + assert tool["outputSchema"]["additionalProperties"] == false + end + + args = %{ + "agent_id" => ctx.worker, + "media_type" => "image/png", + "data_base64" => Base.encode64(@png) + } + + attachment = call(token, @protocol, "attachment_upload", args) |> structured() + assert attachment["agent_id"] == ctx.worker + assert attachment["byte_count"] == byte_size(@png) + assert attachment["media_type"] == "image/png" + assert attachment["sha256"] == Base.encode16(:crypto.hash(:sha256, @png), case: :lower) + assert is_binary(attachment["retention"]) + assert {:ok, _date, 0} = DateTime.from_iso8601(attachment["created_at"]) + refute Map.has_key?(attachment, "path") + refute Map.has_key?(attachment, "data_base64") + assert :offline = Custode.Agents.live_provider(ctx.worker) + + assert call(token, @protocol, "attachment_upload", args) |> structured() == attachment + read_args = %{"attachment_id" => attachment["attachment_id"]} + read = call(token, @protocol, "attachment_read", read_args) |> structured() + assert Map.delete(read, "data_base64") == attachment + assert Base.decode64!(read["data_base64"]) == @png + + prompt = %{ + "agent_id" => ctx.worker, + "prompt" => "", + "attachment_id" => attachment["attachment_id"], + "idempotency_key" => uid("image-only") + } + + receipt = call(token, @protocol, "prompt_agent", prompt) |> text_json() + assert receipt["attachment_id"] == attachment["attachment_id"] + assert receipt["delivered"] == true + assert receipt["duplicate"] == false + stored = Custode.OperatorMessages.get(receipt["message_id"]) + assert stored.attachment_id == attachment["attachment_id"] + assert stored.caller_id == actor.id + assert stored.target_agent_id == ctx.worker + assert stored.prompt =~ "attached image:" + assert [job] = jobs(ctx.worker) + assert job.args["prompt"] =~ "attached image:" + assert job.meta["correlation_id"] == stored.provider_correlation_id + + retry = call(token, @protocol, "prompt_agent", prompt) |> text_json() + assert retry["message_id"] == receipt["message_id"] + assert retry["attachment_id"] == attachment["attachment_id"] + assert retry["duplicate"] == true + assert length(jobs(ctx.worker)) == 1 + + changed = + call(token, @protocol, "attachment_upload", %{ + args + | "data_base64" => Base.encode64(@png <> "changed") + }) + |> structured() + + assert_error( + call(token, @protocol, "prompt_agent", %{ + prompt + | "attachment_id" => changed["attachment_id"] + }) + ) + + assert length(jobs(ctx.worker)) == 1 + + # A busy target queues text and its image as the same durable message. + queued = + call(token, @protocol, "prompt_agent", %{ + prompt + | "prompt" => "inspect next", + "idempotency_key" => uid("queued-image") + }) + |> text_json() + + assert queued["how"] == "queued" + assert queued["attachment_id"] == attachment["attachment_id"] + assert Custode.OperatorMessages.get(queued["message_id"]).prompt =~ "inspect next" + assert length(jobs(ctx.worker)) == 1 + + other_token = Identity.mint(:operator, uid("other-human")) + initialize(other_token, @protocol) + assert_error(call(other_token, @protocol, "attachment_read", read_args)) + + assert_error( + call(other_token, @protocol, "prompt_agent", %{ + prompt + | "idempotency_key" => uid("foreign-image") + }) + ) + + assert_error( + call(token, @protocol, "prompt_agent", %{ + prompt + | "agent_id" => ctx.other, + "idempotency_key" => uid("wrong-target") + }) + ) + + assert :offline = Custode.Agents.live_provider(ctx.other) + + # Legacy optional null keys still mean a new text-only message. + plain = + call(token, @protocol, "prompt_agent", %{ + "agent_id" => ctx.other, + "prompt" => "legacy null idempotency key", + "idempotency_key" => nil + }) + |> text_json() + + assert plain["delivered"] == true + assert plain["attachment_id"] == nil + assert length(jobs(ctx.other)) == 1 + end + + test "upload envelope and refusals stay bounded over HTTP #{@protocol}", ctx do + token = Identity.mint(:operator, uid("attachment-human")) + initialize(token, @protocol) + + args = %{ + "agent_id" => ctx.worker, + "media_type" => "image/png", + "data_base64" => Base.encode64(@png) + } + + # Header validation is deliberate; this is not a full image decoder proof. + large = @png <> :binary.copy(<<0>>, 2_100_000) + large_args = %{args | "data_base64" => Base.encode64(large)} + assert byte_size(Jason.encode!(large_args)) > 2_000_000 + receipt = call(token, @protocol, "attachment_upload", large_args) |> structured() + assert receipt["byte_count"] == byte_size(large) + assert :offline = Custode.Agents.live_provider(ctx.worker) + + oversized = @png <> :binary.copy(<<0>>, 10_000_001 - byte_size(@png)) + + assert_error( + call(token, @protocol, "attachment_upload", %{ + args + | "data_base64" => Base.encode64(oversized) + }) + ) + + for invalid <- ["not base64!", Base.encode64("not an image")] do + assert_error( + call(token, @protocol, "attachment_upload", %{args | "data_base64" => invalid}) + ) + end + + assert_error( + call(token, @protocol, "attachment_upload", %{args | "media_type" => "image/jpeg"}) + ) + + assert_error(call(token, @protocol, "attachment_read", %{"attachment_id" => "missing"})) + + for invalid <- [ + Map.put(args, "path", "/tmp/not-authority.png"), + %{args | "media_type" => "text/plain"}, + %{args | "agent_id" => ""} + ] do + assert_invalid( + post(token, @protocol, "tools/call", %{ + "name" => "attachment_upload", + "arguments" => invalid + }) + ) + end + + assert_invalid( + post(token, @protocol, "tools/call", %{ + "name" => "prompt_agent", + "arguments" => %{ + "agent_id" => ctx.worker, + "prompt" => "must not send", + "attachment_id" => nil + } + }) + ) + + for name <- @tools do + routine_token = Identity.mint(:routine, ctx.caretaker) + initialize(routine_token, @protocol) + refute name in Enum.map(discover(routine_token, @protocol), & &1["name"]) + + arguments = + if name == "attachment_upload", + do: args, + else: %{"attachment_id" => receipt["attachment_id"]} + + assert_refused(call(routine_token, @protocol, name, arguments)) + + assert post(nil, @protocol, "tools/call", %{"name" => name, "arguments" => arguments}).status == + 401 + end + + routine_token = Identity.mint(:routine, ctx.caretaker) + + assert_error( + call(routine_token, @protocol, "prompt_agent", %{ + "agent_id" => ctx.worker, + "prompt" => "do not drop this", + "attachment_id" => receipt["attachment_id"] + }) + ) + + assert Repo.all(from(m in OperatorMessage, where: m.target_agent_id == ^ctx.worker)) == [] + assert jobs(ctx.worker) == [] + assert SnodoTransport.plug_options()["/mcp"].max_body_bytes == 14_000_000 + assert SnodoTransport.plug_options()["/mcp/memory"].max_body_bytes == 2_000_000 + end + end + + defp structured(response) do + assert %{ + "result" => %{ + "isError" => false, + "structuredContent" => value, + "content" => [%{"text" => text}] + } + } = response + + assert Jason.decode!(text) == value + value + end + + defp text_json(response) do + assert %{"result" => %{"isError" => false, "content" => [%{"text" => text}]}} = response + Jason.decode!(text) + end + + defp assert_error(response), do: assert(match?(%{"result" => %{"isError" => true}}, response)) + defp assert_refused(%{"error" => _error}), do: :ok + defp assert_refused(response), do: assert_error(response) + + defp assert_invalid(response) do + assert response.status in [200, 400] + + assert match?(%{"error" => %{"code" => -32_602}}, response.body) or + match?(%{"result" => %{"isError" => true}}, response.body) + end + + defp jobs(agent_id) do + Repo.all( + from(j in Oban.Job, + where: j.worker == "ObanClaude.Agent.Job", + where: fragment("json_extract(?, '$.agent_id')", j.meta) == ^agent_id, + order_by: [asc: j.id] + ) + ) + end + + defp initialize(token, "2025-11-25" = protocol) do + response = + post(token, nil, "initialize", %{ + "protocolVersion" => protocol, + "capabilities" => %{}, + "clientInfo" => %{"name" => "attachment-fixture", "version" => "1"} + }) + + assert response.status == 200 + end + + defp initialize(_token, _protocol), do: :ok + + defp discover(token, protocol, params \\ %{}) do + response = post(token, protocol, "tools/list", params) + assert response.status == 200 + page = response.body["result"] + + case page["nextCursor"] do + nil -> page["tools"] + cursor -> page["tools"] ++ discover(token, protocol, %{"cursor" => cursor}) + end + end + + defp call(token, protocol, name, arguments) do + response = post(token, protocol, "tools/call", %{"name" => name, "arguments" => arguments}) + assert response.status == 200 + response.body + end + + defp post(token, protocol, method, params) do + headers = [{"accept", "application/json, text/event-stream"}] + headers = if token, do: [{"authorization", "Bearer " <> token} | headers], else: headers + headers = if protocol, do: [{"mcp-protocol-version", protocol} | headers], else: headers + {headers, params} = modern_request(protocol, method, headers, params) + + Req.post!(Custode.MCP.url(), + json: %{ + "jsonrpc" => "2.0", + "id" => System.unique_integer([:positive]), + "method" => method, + "params" => params + }, + headers: headers, + retry: false, + receive_timeout: 20_000 + ) + end + + defp modern_request("2026-07-28" = protocol, method, headers, params) do + headers = [{"mcp-method", method} | headers] + + headers = + if method == "tools/call", do: [{"mcp-name", params["name"]} | headers], else: headers + + params = + Map.put(params, "_meta", %{ + "io.modelcontextprotocol/protocolVersion" => protocol, + "io.modelcontextprotocol/clientCapabilities" => %{} + }) + + {headers, params} + end + + defp modern_request(_protocol, _method, headers, params), do: {headers, params} +end diff --git a/test/custode/operator_attachment_messages_test.exs b/test/custode/operator_attachment_messages_test.exs new file mode 100644 index 00000000..7c3963d8 --- /dev/null +++ b/test/custode/operator_attachment_messages_test.exs @@ -0,0 +1,314 @@ +defmodule Custode.OperatorAttachmentMessagesTest do + use ExUnit.Case, async: false + + import Custode.TestHelpers + import Ecto.Query, only: [from: 2] + + alias Custode.{Agents, OperatorMessage, OperatorMessages, Repo, Routine} + alias Custode.Operator.{Actions, Attachment, Attachments} + + @png Base.decode64!( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" + ) + + setup do + routine = routine_fixture!(tmp_workspace!(), %{cron: :manual}) + actor = %{kind: :operator, id: uid("attachment-operator")} + {:ok, attachment} = Attachments.upload(actor, routine.id, "image/png", @png) + %{routine: routine, actor: actor, attachment: attachment} + end + + defp opts(ctx, key \\ "attachment-request") do + [actor: ctx.actor, attachment_id: ctx.attachment.attachment_id, idempotency_key: key] + end + + test "concurrent image-only retries pin one image and deliver once after commit", ctx do + parent = self() + + results = + 1..8 + |> Task.async_stream( + fn _ -> + OperatorMessages.submit(ctx.routine.id, " ", opts(ctx), fn message -> + refute Repo.in_transaction?() + send(parent, {:delivered_image, message}) + {:ok, :queued} + end) + end, + max_concurrency: 8, + timeout: 5_000 + ) + |> Enum.map(fn {:ok, result} -> result end) + + assert_receive {:delivered_image, message} + refute_receive {:delivered_image, _}, 100 + assert Enum.count(results, &match?({:ok, _, :created}, &1)) == 1 + assert Enum.count(results, &match?({:ok, _, :duplicate}, &1)) == 7 + assert message.attachment_id == ctx.attachment.attachment_id + assert message.prompt =~ "attached image: " + assert Repo.get!(OperatorMessage, message.id).attachment_id == message.attachment_id + assert OperatorMessages.public(message).attachment_id == message.attachment_id + assert OperatorMessages.public_summary(message).attachment_id == message.attachment_id + end + + test "changed text or attachment conflicts without another delivery", ctx do + deliver = fn _ -> {:ok, :queued} end + + assert {:ok, _, :created} = + OperatorMessages.submit(ctx.routine.id, "inspect", opts(ctx), deliver) + + assert {:error, :idempotency_conflict} = + OperatorMessages.submit(ctx.routine.id, "publish", opts(ctx), deliver) + + assert {:error, :idempotency_conflict} = + OperatorMessages.submit( + ctx.routine.id, + "inspect", + Keyword.delete(opts(ctx), :attachment_id), + deliver + ) + + assert {:error, :idempotency_conflict} = + OperatorMessages.submit( + ctx.routine.id, + "inspect", + Keyword.put(opts(ctx), :attachment_id, Ecto.UUID.generate()), + deliver + ) + end + + test "text shaped like an attachment fingerprint is not the same request", ctx do + text = "attachment-v1:" <> Jason.encode!(["inspect", ctx.attachment.attachment_id]) + plain_opts = Keyword.delete(opts(ctx), :attachment_id) + + assert {:ok, _, :created} = + OperatorMessages.submit(ctx.routine.id, text, plain_opts, fn _ -> {:ok, :queued} end) + + assert {:error, :idempotency_conflict} = + OperatorMessages.submit(ctx.routine.id, "inspect", opts(ctx), fn _ -> + flunk("different request delivered") + end) + end + + test "a handled text-only conflict does not roll back its caller transaction", ctx do + plain_opts = Keyword.delete(opts(ctx), :attachment_id) + + assert {:ok, _, :created} = + OperatorMessages.submit(ctx.routine.id, "inspect", plain_opts, fn _ -> + {:ok, :queued} + end) + + assert {:ok, :caller_committed} = + Repo.transaction(fn -> + assert {:error, :idempotency_conflict} = + OperatorMessages.submit(ctx.routine.id, "different", plain_opts, fn _ -> + flunk("conflicting delivery") + end) + + :caller_committed + end) + end + + test "the retained receipt survives expired bytes and a workspace move", ctx do + assert {:ok, message, :created} = + OperatorMessages.submit(ctx.routine.id, "inspect", opts(ctx), fn _ -> + {:ok, :queued} + end) + + row = Repo.get_by!(Attachment, attachment_id: ctx.attachment.attachment_id) + Repo.update!(Ecto.Changeset.change(row, deleted_at: DateTime.utc_now())) + File.rm!(row.path) + put_env!(:routines, [%{ctx.routine | workspace: tmp_workspace!()}]) + + assert {:ok, duplicate, :duplicate} = + OperatorMessages.submit(ctx.routine.id, " inspect ", opts(ctx), fn _ -> + flunk("redelivered") + end) + + assert duplicate.message_id == message.message_id + assert duplicate.prompt == message.prompt + + assert {:error, _} = + OperatorMessages.submit(ctx.routine.id, "inspect", opts(ctx, "new-request"), fn _ -> + flunk("delivered expired attachment") + end) + end + + test "invalid ownership, target and delegation create no durable message", ctx do + before = Repo.aggregate(OperatorMessage, :count) + + for {target, actor} <- [ + {ctx.routine.id, %{kind: :operator, id: uid("other-operator")}}, + {uid("wrong-target"), ctx.actor}, + {ctx.routine.id, %{kind: :routine, id: ctx.routine.id}} + ] do + assert {:error, _} = + OperatorMessages.submit( + target, + "inspect", + Keyword.put(opts(ctx), :actor, actor), + fn _ -> flunk("unauthorized delivery") end + ) + end + + assert Repo.aggregate(OperatorMessage, :count) == before + end + + test "attachment admission refuses an outer transaction before provider effects", ctx do + assert {:ok, {:error, :outer_transaction_unsupported}} = + Repo.transaction(fn -> + OperatorMessages.submit(ctx.routine.id, "inspect", opts(ctx), fn _ -> + flunk("delivery before caller commit") + end) + end) + end + + test "concurrent cleanup either expires before admission or preserves accepted bytes", ctx do + for iteration <- 1..8 do + {:ok, image} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png <> to_string(iteration)) + + row = Repo.get_by!(Attachment, attachment_id: image.attachment_id) + old = DateTime.add(DateTime.utc_now(), -90, :day) + Repo.update!(Ecto.Changeset.change(row, inserted_at: old, updated_at: old)) + + request_opts = + opts(ctx, "prune-race-#{iteration}") |> Keyword.put(:attachment_id, image.attachment_id) + + submit = + Task.async(fn -> + receive do + :go -> + OperatorMessages.submit(ctx.routine.id, "inspect", request_opts, fn _ -> + {:ok, :queued} + end) + end + end) + + cleanup = Task.async(fn -> receive do: (:go -> Attachments.prune(30)) end) + send(submit.pid, :go) + send(cleanup.pid, :go) + result = Task.await(submit) + Task.await(cleanup) + + case result do + {:ok, message, :created} -> + assert message.attachment_id == image.attachment_id + assert File.read!(row.path) == @png <> to_string(iteration) + assert Repo.get!(Attachment, row.id).deleted_at == nil + + {:error, :attachment_expired} -> + refute File.exists?(row.path) + + refute Repo.exists?( + from(m in OperatorMessage, where: m.attachment_id == ^image.attachment_id) + ) + end + end + end + + test "an invalid image reference cannot start an offline routine", ctx do + assert {:error, _} = + Actions.message_with_receipt( + ctx.routine.id, + "inspect", + Keyword.put(opts(ctx), :attachment_id, Ecto.UUID.generate()) + ) + + assert {:ok, :offline} = Agents.status(ctx.routine.id) + end + + for provider <- [:claude, :codex], text <- ["", "look at this image"] do + @provider provider + @text text + test "#{provider} accepts #{inspect(text)} plus image offline, paused and busy", ctx do + provider = @provider + routine = ctx.routine |> Map.put(:provider, provider) |> Map.delete(:model) + put_env!(:routines, [routine]) + on_exit(fn -> Agents.stop_agent(routine.id, provider) end) + + assert {:ok, offline, :created} = + Actions.message_with_receipt(routine.id, @text, opts(ctx, "offline")) + + assert offline.delivery == "started" + + job = + eventually(fn -> + jobs = + Repo.all( + from(j in Oban.Job, + where: fragment("json_extract(?, '$.agent_id')", j.meta) == ^routine.id + ) + ) + + assert job = Enum.find(jobs, &(&1.meta["correlation_id"] == offline.message_id)) + job + end) + + assert job.args["prompt"] == offline.prompt + assert offline.prompt =~ "attached image: " + assert File.read!(image_path(offline.prompt)) == @png + finish(provider, job) + assert {:ok, :idle} = Agents.await(routine.id, provider, :idle, 1_000) + assert :ok = Agents.stop_agent(routine.id, provider) + + parent = self() + + config = + routine.id + |> Routine.get() + |> Routine.agent_config(%{}) + |> Keyword.put(:enqueue_fun, fn args, meta -> + send(parent, {:enqueued_attachment, args, meta}) + {:ok, :queued} + end) + + assert {:ok, _} = Agents.start_agent(routine.id, config) + assert :ok = Agents.emergency_pause(routine.id, provider) + assert {:ok, :paused} = Agents.await(routine.id, provider, :paused, 1_000) + + assert {:ok, paused, :created} = + Actions.message_with_receipt(routine.id, @text, opts(ctx, "paused")) + + assert paused.delivery == "resumed" + assert_receive {:enqueued_attachment, paused_args, paused_meta}, 1_000 + assert paused_args["prompt"] == paused.prompt + assert paused_meta["correlation_id"] == paused.message_id + + assert {:ok, queued, :created} = + Actions.message_with_receipt(routine.id, @text, opts(ctx, "busy")) + + assert queued.delivery == "queued" + assert queued.attachment_id == ctx.attachment.attachment_id + refute_receive {:enqueued_attachment, _, _}, 100 + finish(provider, %{meta: paused_meta}) + assert_receive {:enqueued_attachment, queued_args, queued_meta}, 1_000 + assert queued_args["prompt"] == queued.prompt + assert queued_meta["correlation_id"] == queued.message_id + finish(provider, %{meta: queued_meta}) + end + end + + defp image_path(prompt) do + [_, path] = Regex.run(~r/attached image: (.+) -- Read it before answering/, prompt) + path + end + + defp finish(:claude, job) do + result = ObanClaude.Testing.result(result: "done", session_id: Ecto.UUID.generate()) + + ObanClaude.Agent.Job.handle_result( + result, + struct(Oban.Job, Map.merge(%{attempt: 1, max_attempts: 1}, Map.take(job, [:id, :meta]))) + ) + end + + defp finish(:codex, job) do + result = ObanCodex.Testing.result("done", session_id: Ecto.UUID.generate()) + + ObanCodex.Agent.Job.handle_result( + result, + struct(Oban.Job, Map.merge(%{attempt: 1, max_attempts: 1}, Map.take(job, [:id, :meta]))) + ) + end +end diff --git a/test/custode/operator_attachment_storage_test.exs b/test/custode/operator_attachment_storage_test.exs new file mode 100644 index 00000000..7769394e --- /dev/null +++ b/test/custode/operator_attachment_storage_test.exs @@ -0,0 +1,330 @@ +defmodule Custode.OperatorAttachmentStorageTest do + use ExUnit.Case, async: false + + import Custode.TestHelpers + + alias Custode.Operator.{Attachment, Attachments} + alias Custode.{OperatorMessage, Repo, Uploads} + + @png Base.decode64!( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" + ) + + setup do + routine = routine_fixture!(tmp_workspace!()) + actor = %{kind: :operator, id: uid("image-owner")} + %{routine: routine, actor: actor} + end + + test "upload returns owned metadata and read returns exact bytes without paths", ctx do + assert {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + assert image.agent_id == ctx.routine.id + assert image.byte_count == byte_size(@png) + assert image.media_type == "image/png" + assert image.sha256 == Base.encode16(:crypto.hash(:sha256, @png), case: :lower) + assert {:ok, _, _} = DateTime.from_iso8601(image.created_at) + assert image.retention =~ "Active messages" + refute Map.has_key?(image, :path) + refute Map.has_key?(image, :workspace) + refute Map.has_key?(image, :owner_id) + assert {:ok, read} = Attachments.read(ctx.actor, image.attachment_id) + assert Map.delete(read, :data_base64) == image + assert Base.decode64!(read.data_base64) == @png + + row = row(image) + assert Path.basename(row.path) =~ ~r/^attachment-[a-z0-9-]+\.png$/ + assert Path.dirname(row.path) == Path.join(Path.expand(ctx.routine.workspace), "uploads") + end + + test "concurrent identical uploads deduplicate within owner and target", ctx do + results = + 1..6 + |> Task.async_stream( + fn _ -> + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + end, + max_concurrency: 6 + ) + |> Enum.map(fn {:ok, {:ok, image}} -> image.attachment_id end) + + assert length(Enum.uniq(results)) == 1 + + assert {:ok, other} = + Attachments.upload( + %{ctx.actor | id: uid("other")}, + ctx.routine.id, + "image/png", + @png + ) + + refute other.attachment_id in results + assert length(File.ls!(Path.join(ctx.routine.workspace, "uploads"))) == 2 + end + + test "only authenticated operators can upload, retrieve or resolve", ctx do + for actor <- [ + nil, + %{kind: :operator}, + %{kind: :operator, id: ""}, + %{kind: :routine, id: ctx.routine.id} + ] do + assert {:error, :human_identity_required} = + Attachments.upload(actor, ctx.routine.id, "image/png", @png) + + assert {:error, :human_identity_required} = + Attachments.upload_file(actor, ctx.routine.id, "/missing", "image.png") + + assert {:error, :human_identity_required} = Attachments.read(actor, "missing") + + assert {:error, :human_identity_required} = + Attachments.resolve(actor, "missing", ctx.routine.id) + end + end + + test "resolution requires a transaction, exact owner and exact target", ctx do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + + assert {:error, :transaction_required} = + Attachments.resolve(ctx.actor, image.attachment_id, ctx.routine.id) + + assert {:error, :attachment_forbidden} = + Attachments.read(%{ctx.actor | id: "another"}, image.attachment_id) + + assert {:error, :attachment_not_found} = Attachments.read(ctx.actor, Ecto.UUID.generate()) + assert {:error, :invalid_attachment} = Attachments.read(ctx.actor, nil) + + assert {:ok, :resolved} = + Repo.transaction( + fn -> + assert {:error, :attachment_target_mismatch} = + Attachments.resolve(ctx.actor, image.attachment_id, "another") + + assert {:ok, path} = + Attachments.resolve(ctx.actor, image.attachment_id, ctx.routine.id) + + assert path == row(image).path + :resolved + end, + mode: :immediate + ) + end + + test "workspace changes invalidate old references and prevent cross-workspace dedup", ctx do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + put_env!(:routines, [%{ctx.routine | workspace: tmp_workspace!()}]) + + assert {:error, :attachment_workspace_changed} = + Attachments.read(ctx.actor, image.attachment_id) + + assert {:ok, {:error, :attachment_workspace_changed}} = + Repo.transaction( + fn -> + Attachments.resolve(ctx.actor, image.attachment_id, ctx.routine.id) + end, + mode: :immediate + ) + + assert {:ok, new} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + refute new.attachment_id == image.attachment_id + end + + test "bounded format recognition refuses unsupported, mismatched and oversized data", ctx do + assert {:error, :unsupported_media_type} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/svg+xml", "") + + assert {:error, :invalid_image} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/jpeg", @png) + + assert {:error, :invalid_image} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", "") + + assert {:error, :no_such_routine} = + Attachments.upload(ctx.actor, uid("missing"), "image/png", @png) + + bytes = @png <> :binary.copy("x", Attachments.max_bytes()) + + assert {:error, :too_large} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", bytes) + + tmp = Path.join(ctx.routine.workspace, "large.png") + File.write!(tmp, bytes) + + assert {:error, :too_large} = + Attachments.upload_file(ctx.actor, ctx.routine.id, tmp, "photo.png") + end + + test "the upload boundary accepts exactly the configured byte limit", ctx do + bytes = @png <> :binary.copy("x", Attachments.max_bytes() - byte_size(@png)) + assert {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", bytes) + assert image.byte_count == Attachments.max_bytes() + assert {:ok, read} = Attachments.read(ctx.actor, image.attachment_id) + assert Base.decode64!(read.data_base64) == bytes + end + + test "the temporary-file adapter uses bounded identification and server-owned filenames", ctx do + tmp = Path.join(ctx.routine.workspace, "input") + File.write!(tmp, @png) + + assert {:ok, image} = + Attachments.upload_file(ctx.actor, ctx.routine.id, tmp, "../../somewhere.PNG") + + assert Path.basename(row(image).path) =~ ~r/^attachment-/ + + assert {:error, :unsupported_media_type} = + Attachments.upload_file(ctx.actor, ctx.routine.id, tmp, "x.exe") + + File.ln_s!(tmp, tmp <> "-link") + + assert {:error, :unsafe_path} = + Attachments.upload_file(ctx.actor, ctx.routine.id, tmp <> "-link", "x.png") + end + + test "changed, absent and symlinked bytes are refused; reupload creates a fresh reference", + ctx do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + path = row(image).path + File.write!(path, :binary.copy("x", byte_size(@png))) + assert {:error, :attachment_unavailable} = Attachments.read(ctx.actor, image.attachment_id) + assert {:ok, repaired} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + refute repaired.attachment_id == image.attachment_id + assert {:error, :attachment_expired} = Attachments.read(ctx.actor, image.attachment_id) + assert File.read!(path) == :binary.copy("x", byte_size(@png)) + + repaired_path = row(repaired).path + File.rm!(repaired_path) + assert {:error, :attachment_unavailable} = Attachments.read(ctx.actor, repaired.attachment_id) + assert {:ok, replacement} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + refute replacement.attachment_id == repaired.attachment_id + assert {:error, :attachment_expired} = Attachments.read(ctx.actor, repaired.attachment_id) + path = row(replacement).path + File.rm!(path) + outside = Path.join(ctx.routine.workspace, "outside.png") + File.write!(outside, @png) + File.ln_s!(outside, path) + + assert {:error, :attachment_unavailable} = + Attachments.read(ctx.actor, replacement.attachment_id) + + assert Uploads.path(ctx.routine.id, path) == :error + assert {:ok, fresh} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + refute fresh.attachment_id == replacement.attachment_id + assert File.read_link(path) == {:ok, outside} + assert File.read!(outside) == @png + end + + test "a symlinked workspace or upload directory cannot redirect a write", ctx do + outside = tmp_workspace!() + File.ln_s!(outside, Path.join(ctx.routine.workspace, "uploads")) + + assert {:error, :unsafe_path} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + + assert File.ls!(outside) == ["inbox"] + link = Path.join(ctx.routine.workspace, "workspace-link") + File.ln_s!(outside, link) + put_env!(:routines, [%{ctx.routine | workspace: link}]) + + assert {:error, :unsafe_path} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + + assert File.ls!(outside) == ["inbox"] + end + + test "pruning retains active messages and ages terminal links from their latest update", ctx do + for status <- ~w(queued executing waiting_for_input waiting_for_approval) do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png <> status) + old = age(row(image), 90) + message!(image, ctx, status, 90) + Attachments.prune(30) + assert Repo.get!(Attachment, old.id).deleted_at == nil + assert File.exists?(old.path) + end + + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png <> "terminal") + old = age(row(image), 90) + message = message!(image, ctx, "completed", 1) + Attachments.prune(30) + assert File.exists?(old.path) + age(message, 31) + Attachments.prune(30) + refute File.exists?(old.path) + assert Repo.get!(Attachment, old.id).deleted_at + assert {:error, :attachment_expired} = Attachments.read(ctx.actor, image.attachment_id) + + assert {:ok, new} = + Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png <> "terminal") + + refute new.attachment_id == image.attachment_id + end + + test "abandoned uploads expire; tombstones hide files and retry safe unlinking", ctx do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + old = age(row(image), 31) + Attachments.prune(30) + assert Repo.get!(Attachment, old.id).deleted_at + refute File.exists?(old.path) + File.write!(old.path, @png) + assert Uploads.path(ctx.routine.id, old.path) == :error + assert MapSet.member?(Attachments.protected_paths(), old.path) + Attachments.prune(30) + refute File.exists?(old.path) + end + + test "pruning cannot unlink through a substituted directory or a stored path escape", ctx do + {:ok, image} = Attachments.upload(ctx.actor, ctx.routine.id, "image/png", @png) + old = age(row(image), 31) + moved = Path.join(ctx.routine.workspace, "moved-uploads") + File.rename!(Path.dirname(old.path), moved) + File.ln_s!(moved, Path.dirname(old.path)) + Attachments.prune(30) + assert File.exists?(Path.join(moved, Path.basename(old.path))) + assert Repo.get!(Attachment, old.id).deleted_at + + outside = Path.join(ctx.routine.workspace, "outside.png") + File.write!(outside, @png) + old |> Ecto.Changeset.change(path: outside) |> Repo.update!() + Attachments.prune(30) + assert File.read!(outside) == @png + end + + test "legacy prompt paths including spaces remain pinned while the message is active", ctx do + path = Path.join([ctx.routine.workspace, "uploads", "old image.png"]) + + message!( + %{attachment_id: nil}, + ctx, + "waiting_for_input", + 90, + Attachments.compose("inspect", [path]) + ) + + assert MapSet.member?(Attachments.protected_paths(), path) + end + + defp row(image), do: Repo.get_by!(Attachment, attachment_id: image.attachment_id) + + defp age(row, days) do + old = DateTime.add(DateTime.utc_now(), -days, :day) + row |> Ecto.Changeset.change(inserted_at: old, updated_at: old) |> Repo.update!() + end + + defp message!(image, ctx, status, days, prompt \\ "inspect") do + old = DateTime.add(DateTime.utc_now(), -days, :day) + + Repo.insert!(%OperatorMessage{ + message_id: uid("image-message"), + caller_kind: "operator", + caller_id: ctx.actor.id, + transport: "test", + target_agent_id: ctx.routine.id, + attachment_id: image.attachment_id, + idempotency_key: uid("image-request"), + provider_correlation_id: uid("image-correlation"), + prompt_hash: "fixture", + prompt: prompt, + status: status, + inserted_at: old, + updated_at: old + }) + end +end diff --git a/test/custode_web/console_live_test.exs b/test/custode_web/console_live_test.exs index b8db740d..5e59e453 100644 --- a/test/custode_web/console_live_test.exs +++ b/test/custode_web/console_live_test.exs @@ -10,6 +10,7 @@ defmodule CustodeWeb.ConsoleLiveTest do alias Custode.Availability.Collectors.Claude, as: ClaudeUsage alias Custode.GitHub.Cache alias Custode.Operator.Actions + alias Custode.Operator.Attachments alias Custode.Repo alias Custode.Workflow alias Custode.Workflow.BudgetResume @@ -1875,6 +1876,35 @@ defmodule CustodeWeb.ConsoleLiveTest do "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" ) + test "an invalid image does not send its accompanying text or clear the draft", %{ + conn: conn, + sleeper: sleeper + } do + {:ok, view, _html} = live(conn, "/console/#{sleeper.id}") + + view + |> file_input("form[phx-submit=message]", :image, [ + %{name: "invalid.png", content: "not a PNG", type: "image/png"} + ]) + |> render_upload("invalid.png") + + html = + view + |> form("form[phx-submit=message]", %{"text" => "keep this draft"}) + |> render_submit() + + assert html =~ Attachments.error_message(:invalid_image) + assert has_element?(view, "form#message-0") + subject = sleeper.id + refute_push_event(view, "draft:clear", %{subject: ^subject}) + assert :offline = Custode.Agents.live_provider(sleeper.id) + + assert Repo.aggregate( + from(m in Custode.OperatorMessage, where: m.target_agent_id == ^sleeper.id), + :count + ) == 0 + end + test "a chosen image shows as a chip, can be removed, and rides the message as a path", %{conn: conn, sleeper: sleeper} do {:ok, view, _html} = live(conn, "/console/#{sleeper.id}") @@ -1911,6 +1941,20 @@ defmodule CustodeWeb.ConsoleLiveTest do [_, path] = Regex.run(~r/attached image: (\S+)/, turn.args["prompt"]) assert Path.dirname(path) == Path.join(Path.expand(sleeper.workspace), "uploads") assert File.read!(path) == @png + + [message] = + Repo.all(from(m in Custode.OperatorMessage, where: m.target_agent_id == ^sleeper.id)) + + assert is_binary(message.attachment_id) + + assert {:ok, attachment} = + Attachments.read( + %{kind: :operator, id: "operator"}, + message.attachment_id + ) + + assert attachment.agent_id == sleeper.id + assert attachment.data_base64 == Base.encode64(@png) end end diff --git a/test/custode_web/conversation_live_test.exs b/test/custode_web/conversation_live_test.exs index f3290912..49d6ea15 100644 --- a/test/custode_web/conversation_live_test.exs +++ b/test/custode_web/conversation_live_test.exs @@ -2,11 +2,13 @@ defmodule CustodeWeb.ConversationLiveTest do use ExUnit.Case, async: false import Custode.TestHelpers + import Ecto.Query, only: [from: 2] import Phoenix.ConnTest import Phoenix.LiveViewTest alias Custode.{Feed, Gates, OperatorMessage, OperatorMessages, Repo, Routine} alias Custode.Feed.Ingest + alias Custode.Operator.Attachments alias ObanClaude.{Agent, Testing} @endpoint CustodeWeb.Endpoint @@ -81,6 +83,64 @@ defmodule CustodeWeb.ConversationLiveTest do assert has_element?(view, "#message-0 button[type=submit]", "Start and send") end + test "invalid image content preserves the conversation draft without sending its text", %{ + agent: agent, + conn: conn + } do + {:ok, view, _html} = live(conn, "/agents/#{agent.id}/conversation") + + view + |> file_input("form[phx-submit=message]", :image, [ + %{name: "bad.png", content: "not a PNG", type: "image/png"} + ]) + |> render_upload("bad.png") + + html = + view |> form("form[phx-submit=message]", %{"text" => "keep this draft"}) |> render_submit() + + assert html =~ Attachments.error_message(:invalid_image) + assert has_element?(view, "form#message-0") + subject = agent.id + refute_push_event(view, "draft:clear", %{subject: ^subject}) + assert :offline = Custode.Agents.live_provider(agent.id) + assert Repo.all(from(m in OperatorMessage, where: m.target_agent_id == ^agent.id)) == [] + end + + test "an image-only conversation message retains an owned reference", %{ + agent: agent, + conn: conn + } do + png = + Base.decode64!( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" + ) + + {:ok, view, _html} = live(conn, "/agents/#{agent.id}/conversation") + on_exit(fn -> Agent.stop_agent(agent.id) end) + + view + |> file_input("form[phx-submit=message]", :image, [ + %{name: "only.png", content: png, type: "image/png"} + ]) + |> render_upload("only.png") + + view |> form("form[phx-submit=message]", %{"text" => ""}) |> render_submit() + subject = agent.id + assert_push_event(view, "draft:clear", %{subject: ^subject}) + [message] = Repo.all(from(m in OperatorMessage, where: m.target_agent_id == ^agent.id)) + assert is_binary(message.attachment_id) + assert message.prompt =~ "attached image:" + + assert {:ok, attachment} = + Attachments.read( + %{kind: :operator, id: "operator"}, + message.attachment_id + ) + + assert attachment.agent_id == agent.id + assert attachment.data_base64 == Base.encode64(png) + end + test "current input survives a new view and refreshes without starting the agent", %{ agent: agent, conn: conn diff --git a/test/custode_web/feed_thumbnails_test.exs b/test/custode_web/feed_thumbnails_test.exs index 5ba91af3..c8c1db70 100644 --- a/test/custode_web/feed_thumbnails_test.exs +++ b/test/custode_web/feed_thumbnails_test.exs @@ -45,6 +45,19 @@ defmodule CustodeWeb.FeedThumbnailsTest do assert Uploads.url(routine.id, image) == "/agents/#{routine.id}/uploads/shot.png" end + test "refuses a symlinked image and a symlinked uploads directory", %{ + routine: routine, + image: image + } do + File.ln_s!(image, Path.join(Path.dirname(image), "alias.png")) + assert Uploads.path(routine.id, "alias.png") == :error + + moved = Path.join(routine.workspace, "old-uploads") + File.rename!(Path.dirname(image), moved) + File.ln_s!(moved, Path.dirname(image)) + assert Uploads.path(routine.id, "shot.png") == :error + end + test "refuses everything outside it", %{routine: routine, image: image} do assert Uploads.path(routine.id, "../../../etc/passwd") == :error assert Uploads.path(routine.id, "../mix.exs") == :error diff --git a/test/test_helper.exs b/test/test_helper.exs index 716c45fe..5caf5b9c 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -26,6 +26,7 @@ for table <- [ "work_agreement_records", "work_agreements", "operator_messages", + "operator_attachments", "helper_records", "integration_overrides", "integration_requests",