From 8611e4676f60bc08e3a06f78e6bd1c50f5f58b8f Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 16:39:56 -0500 Subject: [PATCH 01/10] docs: add mech storage SDK CAS migration PRD --- ...0005-prd-mech-storage-sdk-cas-migration.md | 256 ++++++++++++++++++ 1 file changed, 256 insertions(+) create mode 100644 tasks/0005-prd-mech-storage-sdk-cas-migration.md diff --git a/tasks/0005-prd-mech-storage-sdk-cas-migration.md b/tasks/0005-prd-mech-storage-sdk-cas-migration.md new file mode 100644 index 0000000..dfaef2a --- /dev/null +++ b/tasks/0005-prd-mech-storage-sdk-cas-migration.md @@ -0,0 +1,256 @@ +# PRD: Supported Mech Storage SDK and CAS Migration + +**Status:** Draft — reviewed, awaiting implementation authorization +**Owner:** AgentDispatch +**Priority:** Critical +**Created:** 2026-08-16 +**Related ledger item:** `task-2026-08-16-001` + +## 1. Problem and outcome + +AgentDispatch's `MechStorage` adapter uses legacy raw NoSQL routes for document get, +put, and delete. The currently installed Mech Storage SDK (`@mech/storage-sdk@0.3.2`) +exposes app-scoped NoSQL operations and a revisioned compare-and-swap (CAS) interface +instead. The adapter also applies partial patches through document replacement routes; +this can erase fields. Inbox lease, acknowledgement, and requeue transitions are +read-then-write operations, so concurrent placements can claim the same message. + +The outcome is a supported, SDK-backed storage adapter that preserves complete records, +uses conditional state transitions for message delivery, handles paged datasets, and is +proven by a real Mech Storage integration canary before production release. + +## 2. Goals + +1. Eliminate AgentDispatch runtime dependencies on unsupported raw single-document + NoSQL endpoints. +2. Use `@mech/storage-sdk` and an explicit application scope for all supported Mech + Storage persistence operations. +3. Make inbox claim, lease renewal, acknowledgement, nack/requeue, and issued-key burn + conflict-safe under concurrent workers. +4. Prevent partial updates from deleting unrelated persisted fields. +5. Paginate all collection scans so correctness does not depend on the first 1,000 rows. +6. Establish repeatable integration validation against a non-production Mech Storage app, + followed by a production canary with no secrets in the repository. + +## 3. Non-goals + +- Changing ADMP's signed-message envelope, SeedID contract, or authentication policy. +- Fixing the separate `brain-msg` false-green hub polling behavior. +- Combining the CLI `nack --extend` payload mismatch into this migration. That is a + small, separately releasable CLI fix (`extend_sec`, not `extend`). +- Rotating or storing production credentials in code, git, task artifacts, or logs. +- Inventing a physical-delete call where the supported SDK/API does not provide one. + +## 4. Users and stories + +### Delivery worker + +As a delivery worker, I can claim exactly one queued message using an atomic lease, so a +second placement does not receive the same message simultaneously. + +### API caller + +As an ADMP client, I can ack, nack, or extend a message lease without erasing message +fields such as payload, sender, recipient, timestamps, or signature metadata. + +### Operator + +As an operator, I can configure Mech Storage using externally supplied endpoint, +application ID, and API key values, and validate the live path using an observable +canary without exposing secrets. + +### Maintainer + +As a maintainer, I can run tests without an ambient production `.env` causing credential +failures or destructive writes. + +## 5. Functional requirements + +### FR-1: SDK boundary + +- Add and pin a compatible `@mech/storage-sdk` dependency in AgentDispatch. +- Construct the SDK from explicit, validated configuration (`baseUrl`, API key, and + application ID). Do not hard-code endpoint, app ID, or credentials. +- Replace generic raw HTTP document access in `src/storage/mech.js` with supported SDK + calls. A narrowly documented compatibility wrapper is permitted only if it calls a + supported SDK method and has contract coverage. +- Fail closed with actionable diagnostics when required Mech configuration is absent or + authentication is rejected; never substitute a memory backend in production. + +### FR-2: Record preservation + +- Every mutation must read the current document/version, merge the permitted patch with + the complete record, and conditionally persist the full next record. +- Reject stale writes as conflicts; do not overwrite a newer revision. +- Define a bounded retry policy for retry-safe conflicts. Retries must reload state and + re-evaluate the transition, rather than replaying stale data. +- Preserve public storage interface return shapes unless a deliberately versioned + interface change is approved. + +### FR-3: Atomic inbox state machine + +- Add storage-level conditional primitives sufficient to implement these transitions: + `queued -> leased`, `leased -> delivered`, `leased -> queued`, and lease expiry + recovery. +- A claim must verify eligibility (recipient/group, status, visibility, and lease + expiry) in the same CAS retry loop that writes the lease. +- Ack and nack must verify the lease owner/token and expected state before transition. +- Lease expiry recovery must not requeue a valid renewed lease. +- Concurrent claim tests must demonstrate that exactly one claimant succeeds for a + message. The loser receives a conflict/no-message result, not the message payload. + +### FR-4: Collections and pagination + +- Replace fixed `limit=1000` scans with an SDK-supported page iterator or cursor loop. +- Apply collection filters as early as the SDK permits and bound resource use with an + explicit maximum/page budget where an unbounded scan is inappropriate. +- Add regression coverage for records beyond the first page for inbox lookup, cleanup, + key lookup, tenant/group listing, and any other collection query used by the adapter. + +### FR-5: Delete and retention contract + +- Before implementation, confirm whether the current Mech Storage service offers a + supported SDK/API delete primitive. +- If it does, add the method to the SDK integration and test authorization and + idempotency. +- If it does not, choose and document one approved alternative before code begins: + (a) tombstone records with a retention job, (b) make logical deletion a supported + service API enhancement, or (c) explicitly change the ADMP data-retention contract. +- Existing raw delete routes must not remain as a hidden fallback. + +### FR-6: Test isolation and validation + +- Unit tests use deterministic in-memory or mocked transport configuration and do not + load live storage credentials by default. +- Add SDK contract tests using a transport stub that asserts app scoping, request shape, + revision propagation, pagination, conflicts, and auth error handling. +- Add an opt-in integration suite requiring explicitly supplied non-production Mech + Storage credentials. It creates an isolated app/namespace or unique test prefix and + cleans it according to the approved retention design. +- Add a post-deploy canary: register two disposable agents, send one signed message, + claim it concurrently, verify one lease winner, ack it, and verify final delivery. + The canary must redact credentials and message payloads in output. + +## 6. Technical design boundaries + +The migration targets the SDK's app-scoped NoSQL and CAS APIs. It must not assume that +legacy endpoints are retained merely because existing code calls them. The adapter may +introduce internal helpers such as `getWithRevision`, `casCreate`, `casReplace`, +`listAllPages`, and message-specific conditional transition helpers. + +The existing generic `updateMessage` interface is insufficient as the only protection +for a delivery claim because eligibility is decided before the write. The storage +boundary therefore needs an explicit atomic claim/transition operation, or an equivalent +adapter-owned CAS loop that owns both eligibility evaluation and persistence. + +No data migration is assumed. Before rollout, inspect a representative non-production +dataset for document IDs, collection names, `_rev` semantics, and fields used by the +adapter. Production data requires a read-only compatibility assessment and backup/export +approval before any write-path rollout. + +## 7. Delivery plan and gates + +### Phase A — design and dependency verification + +1. Confirm the SDK/API delete capability and `_rev` conflict response contract with the + Mech Storage owner or authoritative SDK tests. +2. Map every current adapter method to a supported SDK method or a planned logical + retention behavior. +3. Specify atomic storage interface additions and error mapping. + +**Gate A:** human review of retention choice and interface design; no implementation +until this is accepted. + +### Phase B — implementation in a dedicated feature worktree + +1. Add the SDK dependency and configuration validation. +2. Rewrite read/list/create/update paths with complete-document CAS semantics. +3. Implement paged collection iterators and atomic message transition primitives. +4. Add unit, contract, concurrency, and opt-in integration coverage. + +**Gate B:** relevant Bun tests pass; no unsupported raw route remains; adversarial review +has no unaddressed critical finding. + +### Phase C — staging and production readiness + +1. Configure non-production credentials outside git and run the integration suite. +2. Repair/rotate the separately managed production Mech Storage credential only through + approved secret management; confirm read/write health. +3. Run the signed-message concurrency canary in staging, then production. + +**Gate C:** explicit human authorization for production configuration/auth changes, +successful canary evidence, and pre-push/PR review gates complete. + +### Phase D — release and observation + +1. Deploy with a documented rollback path. +2. Monitor storage authentication errors, CAS conflicts, duplicate claims, lease age, + and canary results. +3. Roll back if delivery duplication, field loss, authorization failures, or conflict + rates exceed agreed thresholds. + +## 8. Acceptance criteria + +- `src/storage/mech.js` makes no raw calls to legacy single-document NoSQL routes. +- All supported mutations retain unrelated record fields and honor current revisions. +- A concurrency test with at least two simultaneous claim attempts yields one winner and + one non-delivery result, repeatedly. +- Ack, nack, extension, and expired-lease recovery reject stale lease state. +- Adapter collection operations required for correctness traverse more than one page. +- The normal unit suite passes without valid live storage credentials. +- The opt-in non-production SDK integration suite passes with externally supplied + credentials. +- The post-deploy canary completes registration, signed send, single claim, ack, and + final inbox verification without leaked secrets. +- The retention/delete decision is implemented and documented; no unsupported fallback + remains. + +## 9. Risks and mitigations + +| Risk | Mitigation | +| --- | --- | +| SDK lacks deletion | Make retention design a blocking Gate A decision; coordinate an SDK/service enhancement if required. | +| CAS migration changes storage shape | Use a representative non-production compatibility assessment and protected rollout. | +| Current credential is invalid | Treat credential repair as a separately authorized operational change; never embed a replacement. | +| CAS conflicts increase under load | Instrument conflicts, use bounded re-read/re-evaluate retries, and test contention explicitly. | +| Broad adapter rewrite regresses uncommon paths | Method-by-method mapping, contract tests, pagination tests, and canary coverage. | + +## 10. Metrics + +- Zero legacy raw single-document storage calls in runtime source. +- Zero duplicate successful claims in concurrency tests and canaries. +- Zero partial-update field-loss regressions in contract tests. +- 100% of required adapter query paths covered by pagination tests. +- Successful signed-message lifecycle canary before and after production deployment. + +## 11. Open decisions + +1. What is the authoritative supported delete/retention API for the current Mech Storage + version? +2. Does the service guarantee CAS atomicity at the document scope under concurrent + requests, and what exact error shape indicates a conflict? +3. What non-production Mech app/namespace and secret-management path are approved for + integration testing? +4. What retention period applies to tombstoned ADMP messages, agents, and issued keys if + physical delete is unavailable? + +## 12. Adversarial review + +**Proposal challenged:** replace the raw adapter with SDK/CAS and use it to make ADMP +delivery safe under concurrency. + +**Strongest objections:** + +- A full rewrite without a delete decision can strand data or quietly preserve an + unsupported fallback. +- CAS alone is not enough if claim eligibility is evaluated outside the conditional + transition. +- A green memory-backend suite does not prove real SDK authentication, app scope, or + revision behavior. +- Repairing invalid production credentials alongside code changes would make root cause + and rollback ambiguous. + +**Verdict:** proceed with planning only. Implementation is **paused at Gate A** until the +delete/retention contract, SDK CAS error semantics, and non-production integration +environment are confirmed. This is a safety gate, not a cancellation of the migration. + From c4895f3463427e3dcbf037414caa49a90ff4da67 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 16:44:35 -0500 Subject: [PATCH 02/10] docs: add mech storage migration task list --- ...0005-prd-mech-storage-sdk-cas-migration.md | 174 ++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md diff --git a/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md b/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md new file mode 100644 index 0000000..779f524 --- /dev/null +++ b/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md @@ -0,0 +1,174 @@ +# Implementation Tasks: Mech Storage SDK and CAS Migration + +**PRD:** `tasks/0005-prd-mech-storage-sdk-cas-migration.md` +**Status:** Ready for implementation, subject to Task 1.0 decision gate +**Planning branch:** `docs/mech-storage-sdk-cas-migration` + +## Relevant Files + +- `package.json` — add the pinned Mech Storage SDK and opt-in test commands. +- `bun.lock` — lock the SDK dependency. +- `src/storage/index.js` — backend selection and configuration boundary. +- `src/storage/mech.js` — current ignored local adapter; must be reviewed and added to + the feature branch before it can be released. +- `src/storage/memory.js` — reference storage contract and test double. +- `src/services/inbox.service.js` — pull, ack, nack, lease expiry, and purge state + transitions that require conditional operations. +- `src/routes/inbox.js` — request validation and mapping of conflict/no-message results. +- `src/server.test.js` — existing end-to-end behavior coverage; run with the memory + backend only unless an integration test explicitly opts in. +- `tests/storage/mech.contract.test.js` (new) — mocked SDK/transport coverage for scope, + revisions, pagination, error mapping, and full-record updates. +- `tests/storage/mech.concurrency.test.js` (new) — competing claim, lease renewal, + ack/nack, and expired-lease race coverage. +- `tests/storage/mech.integration.test.js` (new) — opt-in non-production SDK suite; + requires external credentials and an isolated namespace. +- `scripts/canary-mech-storage.js` (new) — redacted staging/production signed-message + lifecycle canary. +- `.env.example` — document variable names only; never real values. +- `README.md` and `RUNBOOK.md` — supported storage configuration, test commands, + observability, canary, rollback, and retention behavior. +- `memory/daily/2026-08-16.md` — record completion and durable operational findings. + +## Task Ordering and Dependencies + +`1.0` blocks all code changes because delete/retention and CAS error semantics define +the storage interface. `2.0` blocks implementation because the adapter currently exists +only as an ignored local file and must be deliberately reviewed into the feature branch. +`3.0` establishes the SDK adapter primitives used by `4.0` inbox transitions and `5.0` +retention. `6.0` validates those layers, then `7.0` performs the separately authorized +non-production/production rollout. + +## Tasks + +- [ ] 1.0 Confirm the supported Mech Storage contract and approve retention + - [ ] 1.1 Run the SDK's authoritative contract tests or obtain authoritative Mech + Storage documentation for app-scoped CAS get/create/replace, `_rev` propagation, + conflict response shape, pagination cursor shape, and retry safety. + - [ ] 1.2 Confirm whether a supported delete operation exists in the deployed Mech + Storage version. Do not infer support from legacy raw routes. + - [ ] 1.3 If delete is unavailable, obtain explicit owner approval for tombstones plus + retention, or create a Mech Storage enhancement work order; define per-collection + retention periods and cleanup authorization. + - [ ] 1.4 Provision or identify a non-production app/namespace and secret-management + path. Record variable names and access procedure only; never record credentials. + - [ ] 1.5 Write the signed decision and SDK behavior evidence in the PRD/runbook. + - [ ] 1.6 Human approval gate: accept the retention choice and CAS interface before + proceeding to Task 2.0. + +- [ ] 2.0 Establish a clean, reviewable Mech adapter baseline + - [ ] 2.1 Create `feat/mech-storage-sdk-cas` from current `origin/main` in a separate + worktree; do not work in the shared dirty checkout. + - [ ] 2.2 Compare the ignored local `src/storage/mech.js` with `src/storage/memory.js` + and enumerate every required storage-interface method and collection. + - [ ] 2.3 Add the intended adapter source to the feature branch deliberately, without + copying embedded credentials, temporary canaries, or unsupported raw endpoint code. + - [ ] 2.4 Add a storage-interface parity test or method inventory so future backends + cannot silently omit an ADMP operation. + - [ ] 2.5 Verify `STORAGE_BACKEND=memory` remains the deterministic default for unit + tests and that `STORAGE_BACKEND=mech` fails closed on missing configuration. + +- [ ] 3.0 Replace legacy document access with SDK/CAS primitives + - [ ] 3.1 Add a pinned, compatibility-tested `@mech/storage-sdk` dependency and + update `bun.lock`. + - [ ] 3.2 Construct the SDK from configurable base URL, application ID, and API key; + validate all required values and preserve safe error messages. + - [ ] 3.3 Remove the adapter's generic raw `fetch` request helper and legacy + single-document get/put/delete route usage. + - [ ] 3.4 Implement app-scoped create, read-with-revision, CAS create/replace, and + normalized error mapping helpers using supported SDK calls. + - [ ] 3.5 Make every update read the current record, merge permitted fields, and CAS + write the full next record. Use a bounded re-read/re-evaluate retry only for + explicitly retry-safe conflicts. + - [ ] 3.6 Implement a cursor/page iterator and replace fixed first-page collection + scans for agents, tenants, messages/inbox, groups, issued keys, outbox, and round + tables. + - [ ] 3.7 Add contract tests for SDK headers/app scope, document shape, `_rev`, 404, + authentication failure, conflicts, complete-record preservation, and multi-page + iteration. + +- [ ] 4.0 Make message delivery and key consumption conditional + - [ ] 4.1 Add explicit storage operations for conditional message claim, conditional + ack, conditional nack/requeue, conditional lease extension, and lease-expiry + recovery. Do not expose a generic stale patch as the sole concurrency defense. + - [ ] 4.2 Move recipient/status/TTL/lease eligibility evaluation inside the CAS retry + loop used by claim and transition operations. + - [ ] 4.3 Update `InboxService.pull` to use atomic claim semantics for normal and + auto-ack paths while preserving FIFO intent, `retain_until_acked`, and ephemeral + behavior. + - [ ] 4.4 Update ack, nack, purge, and expiry flows to reject stale state, wrong + recipient, and expired/renewed lease transitions consistently. + - [ ] 4.5 Make single-use issued-key consumption conflict-safe so concurrent redemption + cannot succeed twice. + - [ ] 4.6 Map storage conflict/no-message outcomes in routes to stable, documented ADMP + API responses without leaking internal revision values. + - [ ] 4.7 Add concurrency tests for two pullers, renew-versus-expiry, ack-versus-nack, + auto-ack races, and two single-use-key consumers. Repeat contested tests enough to + detect timing-sensitive duplicate claims. + +- [ ] 5.0 Implement approved retention and removal behavior + - [ ] 5.1 Implement the approved SDK delete operation, or add explicit tombstone state + and retention metadata for every collection that currently deletes records. + - [ ] 5.2 Update cleanup, expired-message, ephemeral-purge, domain-config, agent, + tenant, group, and key paths to use the approved behavior; remove all raw delete + fallbacks. + - [ ] 5.3 Ensure list/read operations consistently exclude tombstones where the public + ADMP contract requires absence. + - [ ] 5.4 Add idempotency, authorization, retention-expiry, and pagination tests for + the chosen behavior. + - [ ] 5.5 Document retention period, cleanup trigger, backup/export prerequisite, and + restoration implications in the runbook. + +- [ ] 6.0 Add isolated validation and operational documentation + - [ ] 6.1 Ensure normal tests never load ambient live Mech credentials; use explicit + memory/mock configuration and fail if integration credentials are absent. + - [ ] 6.2 Implement the opt-in non-production integration suite using an isolated app + or unique namespace, externally injected credentials, and approved cleanup. + - [ ] 6.3 Implement a redacted canary that registers disposable agents, sends a signed + message, runs concurrent pull, verifies one winner, acks, and checks final state. + - [ ] 6.4 Add configuration guidance to `.env.example`; document integration test, + canary, metrics, alerts, rollback, and retention in `README.md`/`RUNBOOK.md`. + - [ ] 6.5 Run the relevant Bun test suite, SDK contract suite, concurrency suite, and + opt-in integration suite. Record commands and sanitized outcomes. + +- [ ] 7.0 Review, release, and verify + - [ ] 7.1 Run adversarial review on the implemented change, specifically challenging + CAS atomicity, field-loss protection, retry behavior, data retention, and test + isolation. + - [ ] 7.2 Run the repository pre-push review (including secret scanning) and address + all blocking findings. + - [ ] 7.3 Open a PR with the PRD, task list, retention decision, migration/rollback + plan, and exact validation evidence. Request human review for configuration/auth + changes. + - [ ] 7.4 After each push, obtain code review and run the PR review loop; do not merge + on CI-only evidence. + - [ ] 7.5 Deploy to the approved non-production environment, run the canary, and + inspect authentication errors, CAS conflicts, duplicate claims, and lease age. + - [ ] 7.6 Obtain explicit authorization before production secret/configuration changes, + then deploy and run the redacted production canary. + - [ ] 7.7 If delivery duplication, field loss, authorization failure, or unacceptable + conflict rate occurs, execute the documented rollback and preserve incident evidence. + - [ ] 7.8 After merge, run full local validation, update the task checklist and daily + memory, and promote durable storage lessons to `memory/MEMORY.md`. + +## Full-list Adversarial Review + +**Challenge:** The plan could be mistaken for authorization to rewrite an ignored local +adapter and repair production credentials simultaneously. + +**Findings incorporated:** + +1. Task 1.0 is a hard decision gate: no delete behavior is assumed, and no production + credential is requested or stored. +2. Task 2.0 treats the ignored adapter as untrusted input requiring intentional, + reviewed intake into a clean branch. +3. Task 4.0 requires an atomic storage operation that owns eligibility plus mutation; + merely changing `updateMessage` to CAS would still race. +4. Task 6.0 separates mocked/normal tests from opt-in real-storage tests, preventing a + false-green memory suite. +5. Task 7.0 requires human approval for production configuration/auth changes and a + rollback-tested canary before declaring success. + +**Verdict:** approved as an implementation checklist. Begin at Task 1.1 to gather the +required evidence; do not begin Task 2.0 or any adapter rewrite until Gate 1.6 has +explicit human approval. From cb7dbf8617025a128c3c5a34cbbcf320badf1d91 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 16:54:44 -0500 Subject: [PATCH 03/10] docs: harden mech storage migration plan --- ...0005-prd-mech-storage-sdk-cas-migration.md | 42 +++++++++++++-- ...0005-prd-mech-storage-sdk-cas-migration.md | 52 ++++++++++++++----- 2 files changed, 76 insertions(+), 18 deletions(-) diff --git a/tasks/0005-prd-mech-storage-sdk-cas-migration.md b/tasks/0005-prd-mech-storage-sdk-cas-migration.md index dfaef2a..1ad2cb0 100644 --- a/tasks/0005-prd-mech-storage-sdk-cas-migration.md +++ b/tasks/0005-prd-mech-storage-sdk-cas-migration.md @@ -31,6 +31,8 @@ proven by a real Mech Storage integration canary before production release. 5. Paginate all collection scans so correctness does not depend on the first 1,000 rows. 6. Establish repeatable integration validation against a non-production Mech Storage app, followed by a production canary with no secrets in the repository. +7. Fence every delivery lease with an opaque per-claim capability, so stale placements + cannot mutate a newer lease held by the same agent identity. ## 3. Non-goals @@ -95,6 +97,12 @@ failures or destructive writes. - A claim must verify eligibility (recipient/group, status, visibility, and lease expiry) in the same CAS retry loop that writes the lease. - Ack and nack must verify the lease owner/token and expected state before transition. +- Generate a cryptographically strong opaque `lease_id` or token on every successful + claim; persist it with the lease, return it only to the claimant, require it for ack, + nack/requeue, and extension, and clear it on terminal/requeued state. +- Define a versioned adoption path: ship pull-token support and updated CLI/SDK clients, + measure adoption, then enforce token-required mutation. During transition, document + the exact safety limitation of tokenless clients. - Lease expiry recovery must not requeue a valid renewed lease. - Concurrent claim tests must demonstrate that exactly one claimant succeeds for a message. The loser receives a conflict/no-message result, not the message payload. @@ -143,6 +151,12 @@ for a delivery claim because eligibility is decided before the write. The storag boundary therefore needs an explicit atomic claim/transition operation, or an equivalent adapter-owned CAS loop that owns both eligibility evaluation and persistence. +The adapter currently exists as a gitignored local overlay and is absent from +`origin/main`. Before implementation, the owner must choose either to version and +package the reviewed adapter in AgentDispatch or to make the overlay repository the +explicit implementation/release target with a versioned parity contract. A local ignored +file is not a release artifact. + No data migration is assumed. Before rollout, inspect a representative non-production dataset for document IDs, collection names, `_rev` semantics, and fields used by the adapter. Production data requires a read-only compatibility assessment and backup/export @@ -154,9 +168,15 @@ approval before any write-path rollout. 1. Confirm the SDK/API delete capability and `_rev` conflict response contract with the Mech Storage owner or authoritative SDK tests. -2. Map every current adapter method to a supported SDK method or a planned logical +2. Approve the adapter ownership and packaging model. +3. Map every current adapter method to a supported SDK method or a planned logical retention behavior. -3. Specify atomic storage interface additions and error mapping. +4. Specify atomic storage interface additions, opaque lease-token schema, ADMP/CLI + compatibility rollout, and error mapping. +5. Collect sanitized non-production legacy-record fixtures and prove SDK read, first + CAS write, paginated list, and rollback-read compatibility for every collection. +6. Define authoritative-time or bounded clock-skew behavior for lease expiry, and + compensation/repair for multi-document issued-key indexing. **Gate A:** human review of retention choice and interface design; no implementation until this is accepted. @@ -196,6 +216,10 @@ successful canary evidence, and pre-push/PR review gates complete. - A concurrency test with at least two simultaneous claim attempts yields one winner and one non-delivery result, repeatedly. - Ack, nack, extension, and expired-lease recovery reject stale lease state. +- A stale lease token cannot ack, nack, extend, or requeue a later lease, including when + both placements use the same agent identity. +- The adapter ownership/packaging model is committed and deployable; no ignored local + source is required at runtime. - Adapter collection operations required for correctness traverse more than one page. - The normal unit suite passes without valid live storage credentials. - The opt-in non-production SDK integration suite passes with externally supplied @@ -214,6 +238,8 @@ successful canary evidence, and pre-push/PR review gates complete. | Current credential is invalid | Treat credential repair as a separately authorized operational change; never embed a replacement. | | CAS conflicts increase under load | Instrument conflicts, use bounded re-read/re-evaluate retries, and test contention explicitly. | | Broad adapter rewrite regresses uncommon paths | Method-by-method mapping, contract tests, pagination tests, and canary coverage. | +| Stale placement mutates a renewed lease | Per-claim opaque lease token, CAS predicates, staged client adoption, and stale-token tests. | +| Mixed raw/CAS writers during rollout | Writer freeze/drain and legacy-record compatibility/rollback preflight. | ## 10. Metrics @@ -233,6 +259,10 @@ successful canary evidence, and pre-push/PR review gates complete. integration testing? 4. What retention period applies to tombstoned ADMP messages, agents, and issued keys if physical delete is unavailable? +5. Which repository owns the deployable Mech adapter, and how is its version included in + the production artifact? +6. What API-version/adoption policy will make lease tokens mandatory without concealing + stale-worker risk in legacy clients? ## 12. Adversarial review @@ -245,12 +275,14 @@ delivery safe under concurrency. unsupported fallback. - CAS alone is not enough if claim eligibility is evaluated outside the conditional transition. +- CAS alone is also not a client fence: a per-claim token is required to stop an old + placement of the same agent from mutating a newly claimed lease. - A green memory-backend suite does not prove real SDK authentication, app scope, or revision behavior. - Repairing invalid production credentials alongside code changes would make root cause and rollback ambiguous. **Verdict:** proceed with planning only. Implementation is **paused at Gate A** until the -delete/retention contract, SDK CAS error semantics, and non-production integration -environment are confirmed. This is a safety gate, not a cancellation of the migration. - +delete/retention contract, SDK CAS error semantics, adapter ownership, lease-token +rollout, and non-production integration environment are confirmed. This is a safety +gate, not a cancellation of the migration. diff --git a/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md b/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md index 779f524..379a151 100644 --- a/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md +++ b/tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md @@ -32,8 +32,8 @@ ## Task Ordering and Dependencies -`1.0` blocks all code changes because delete/retention and CAS error semantics define -the storage interface. `2.0` blocks implementation because the adapter currently exists +`1.0` blocks all code changes because delete/retention, lease fencing, adapter ownership, +and CAS error semantics define the storage interface. `2.0` blocks implementation because the adapter currently exists only as an ignored local file and must be deliberately reviewed into the feature branch. `3.0` establishes the SDK adapter primitives used by `4.0` inbox transitions and `5.0` retention. `6.0` validates those layers, then `7.0` performs the separately authorized @@ -53,11 +53,23 @@ non-production/production rollout. - [ ] 1.4 Provision or identify a non-production app/namespace and secret-management path. Record variable names and access procedure only; never record credentials. - [ ] 1.5 Write the signed decision and SDK behavior evidence in the PRD/runbook. - - [ ] 1.6 Human approval gate: accept the retention choice and CAS interface before + - [ ] 1.6 Decide and document the deployable adapter owner: version/package it in + AgentDispatch, or name the overlay repository and its parity/release contract. + - [ ] 1.7 Define the opaque per-claim `lease_id`/token record schema, generation, + storage, clearing, CAS predicates, error response, and client-facing API fields. + - [ ] 1.8 Approve a staged API/CLI rollout for mandatory lease tokens. Define legacy + tokenless behavior, adoption metrics, enforcement date, and compatibility tests. + - [ ] 1.9 Collect sanitized non-production legacy-record fixtures for every collection; + prove SDK read, first CAS mutation, pagination, and rollback-read compatibility. + - [ ] 1.10 Decide storage/server time or a bounded clock-skew policy for expiry, and + specify issued-key primary/index compensation, repair, and idempotency semantics. + - [ ] 1.11 Human approval gate: accept retention, adapter ownership, lease-token + rollout, CAS, legacy-data, and time/index decisions before proceeding to Task 2.0. - [ ] 2.0 Establish a clean, reviewable Mech adapter baseline - - [ ] 2.1 Create `feat/mech-storage-sdk-cas` from current `origin/main` in a separate + - [ ] 2.1 Create `feat/mech-storage-sdk-cas` in the approved adapter-owner repository + from current `origin/main` (or the approved overlay baseline) in a separate worktree; do not work in the shared dirty checkout. - [ ] 2.2 Compare the ignored local `src/storage/mech.js` with `src/storage/memory.js` and enumerate every required storage-interface method and collection. @@ -91,18 +103,23 @@ non-production/production rollout. - [ ] 4.1 Add explicit storage operations for conditional message claim, conditional ack, conditional nack/requeue, conditional lease extension, and lease-expiry recovery. Do not expose a generic stale patch as the sole concurrency defense. - - [ ] 4.2 Move recipient/status/TTL/lease eligibility evaluation inside the CAS retry + - [ ] 4.2 Generate a fresh opaque lease token during claim; return it in pull, require + it for ack/nack/extension, and make token/state/revision part of every CAS predicate. + - [ ] 4.3 Version the inbox API and CLI/SDK client contract for token rollout. Test the + documented legacy mode, token-capable mode, adoption telemetry, and enforcement mode. + - [ ] 4.4 Move recipient/status/TTL/lease eligibility evaluation inside the CAS retry loop used by claim and transition operations. - - [ ] 4.3 Update `InboxService.pull` to use atomic claim semantics for normal and + - [ ] 4.5 Update `InboxService.pull` to use atomic claim semantics for normal and auto-ack paths while preserving FIFO intent, `retain_until_acked`, and ephemeral behavior. - - [ ] 4.4 Update ack, nack, purge, and expiry flows to reject stale state, wrong + - [ ] 4.6 Update ack, nack, purge, and expiry flows to reject stale state, wrong recipient, and expired/renewed lease transitions consistently. - - [ ] 4.5 Make single-use issued-key consumption conflict-safe so concurrent redemption + - [ ] 4.7 Make single-use issued-key consumption conflict-safe so concurrent redemption cannot succeed twice. - - [ ] 4.6 Map storage conflict/no-message outcomes in routes to stable, documented ADMP + - [ ] 4.8 Map storage conflict/no-message outcomes in routes to stable, documented ADMP API responses without leaking internal revision values. - - [ ] 4.7 Add concurrency tests for two pullers, renew-versus-expiry, ack-versus-nack, + - [ ] 4.9 Add concurrency tests for two pullers, old-token-after-reclaim, + renew-versus-expiry under clock skew, ack-versus-nack, auto-ack races, and two single-use-key consumers. Repeat contested tests enough to detect timing-sensitive duplicate claims. @@ -126,12 +143,21 @@ non-production/production rollout. or unique namespace, externally injected credentials, and approved cleanup. - [ ] 6.3 Implement a redacted canary that registers disposable agents, sends a signed message, runs concurrent pull, verifies one winner, acks, and checks final state. - - [ ] 6.4 Add configuration guidance to `.env.example`; document integration test, + - [ ] 6.4 Standardize a validated canonical storage configuration schema, with tested + legacy aliases only where deliberately retained; update `.env.example`, scripts, + diagnostics, and docs together. + - [ ] 6.5 Add configuration guidance to `.env.example`; document integration test, canary, metrics, alerts, rollback, and retention in `README.md`/`RUNBOOK.md`. - - [ ] 6.5 Run the relevant Bun test suite, SDK contract suite, concurrency suite, and + - [ ] 6.6 Run the relevant Bun test suite, SDK contract suite, concurrency suite, and opt-in integration suite. Record commands and sanitized outcomes. - [ ] 7.0 Review, release, and verify + - [ ] 7.0a Ship the independently tracked CLI `nack --extend` compatibility correction + (`extend_sec`) with a request-shape regression test and published-artifact validation + before relying on extension canary coverage. + - [ ] 7.0b Before cutover, inventory representative production records read-only, block + mixed raw/CAS writers with a rollout flag or write freeze, drain/expire active leases, + and prove rollback can read records containing lease-token fields. - [ ] 7.1 Run adversarial review on the implemented change, specifically challenging CAS atomicity, field-loss protection, retry behavior, data retention, and test isolation. @@ -170,5 +196,5 @@ adapter and repair production credentials simultaneously. rollback-tested canary before declaring success. **Verdict:** approved as an implementation checklist. Begin at Task 1.1 to gather the -required evidence; do not begin Task 2.0 or any adapter rewrite until Gate 1.6 has +required evidence; do not begin Task 2.0 or any adapter rewrite until Gate 1.11 has explicit human approval. From 8b7ab16e4bb7cdf9afd1e7cbe9f5dc82574e93a1 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:24:52 -0500 Subject: [PATCH 04/10] feat(storage): migrate mech adapter and fence inbox leases - Use the supported Mech Storage SDK with CAS, pagination, and tombstones - Add atomic claims and lease-token fencing across HTTP and CLI Implements tasks 2.0 through 4.0 from tasks/tasks-0005-prd-mech-storage-sdk-cas-migration.md --- .env.example | 11 +- .gitignore | 3 - bun.lock | 3 + cli/README.md | 4 +- cli/src/commands/ack.ts | 7 +- cli/src/commands/nack.ts | 9 +- cli/src/output.ts | 7 +- docs/AGENT-GUIDE.md | 7 +- docs/API-REFERENCE.md | 7 +- docs/CLI-REFERENCE.md | 4 +- package.json | 1 + src/routes/inbox.js | 18 +- src/server.test.js | 80 ++- src/services/inbox.service.js | 99 +-- src/storage/index.js | 15 +- src/storage/mech.js | 1116 +++++++++++++++++++++++++++++++++ src/storage/mech.test.js | 93 +++ src/storage/memory.js | 72 ++- 18 files changed, 1466 insertions(+), 90 deletions(-) create mode 100644 src/storage/mech.js create mode 100644 src/storage/mech.test.js diff --git a/.env.example b/.env.example index 5005b83..0148153 100644 --- a/.env.example +++ b/.env.example @@ -23,9 +23,18 @@ MAX_MESSAGES_PER_AGENT=1000 # Storage Backend # "memory" — in-process Map, no persistence (default for development) -# Custom backends: implement the interface in src/storage/memory.js +# "mech" — persistent app-scoped Mech Storage via @mech/storage-sdk STORAGE_BACKEND=memory +# Required only when STORAGE_BACKEND=mech. Keep real values in secret management; +# never commit them. MECH_STORAGE_* names are accepted as legacy aliases. +MECH_BASE_URL=https://storage.mechdna.net +MECH_APP_ID= +MECH_API_KEY= +# Lease tokens are required by default. Set to "compat" only during an explicitly +# monitored client migration; tokenless acknowledgement is not multi-placement safe. +LEASE_TOKEN_MODE=required + # Registration Policy # "open" — agents auto-approved on registration # "approval_required" — agents start pending, require admin approval diff --git a/.gitignore b/.gitignore index d2c28e8..2fccd0c 100644 --- a/.gitignore +++ b/.gitignore @@ -71,8 +71,5 @@ PR-*-MERGE-READINESS-FINAL.md PR_DESCRIPTION.md MECH-PERFORMANCE-ANALYSIS.md -# Proprietary storage backend -src/storage/mech.js - # Local settings .claude/settings.local.json diff --git a/bun.lock b/bun.lock index 078c4e2..4716d0b 100644 --- a/bun.lock +++ b/bun.lock @@ -5,6 +5,7 @@ "": { "name": "agent-dispatch", "dependencies": { + "@mech/storage-sdk": "0.3.2", "cors": "^2.8.5", "dotenv": "^17.2.3", "express": "^4.18.2", @@ -24,6 +25,8 @@ }, }, "packages": { + "@mech/storage-sdk": ["@mech/storage-sdk@0.3.2", "https://registry.mechdna.net/npm/@mech/storage-sdk/-/storage-sdk-0.3.2.tgz", {}, "sha512-hb7zXI8ckc2G9/xfjvnCRy2EBYJOiqLFnvlqRx6F9lfJOTsLfMeq3qD9+ePugfhUkXX9ODOyyHlbDH+f3DdGgw=="], + "@noble/hashes": ["@noble/hashes@1.8.0", "", {}, "sha512-jCs9ldd7NwzpgXDIf6P3+NrHh9/sD6CQdxHyjQI+h/6rDNo88ypBxxz45UDuZHz9r3tNz7N/VInSVoVdtXEI4A=="], "@paralleldrive/cuid2": ["@paralleldrive/cuid2@2.3.1", "", { "dependencies": { "@noble/hashes": "^1.1.5" } }, "sha512-XO7cAxhnTZl0Yggq6jOgjiOHhbgcO4NqFqwSmQpjK3b6TEE6Uj/jfSk6wzYyemh3+I0sHirKSetjQwn5cZktFw=="], diff --git a/cli/README.md b/cli/README.md index b2132f0..b3a4da4 100644 --- a/cli/README.md +++ b/cli/README.md @@ -22,8 +22,8 @@ admp send --to analyst-agent --subject task.request --body '{"action":"summarize # 3. Pull your next incoming message (leases it) admp pull -# 4. Acknowledge successful processing -admp ack +# 4. Acknowledge successful processing with the lease token returned by pull +admp ack --lease-token ``` ## Commands diff --git a/cli/src/commands/ack.ts b/cli/src/commands/ack.ts index 379d17a..bf24d53 100644 --- a/cli/src/commands/ack.ts +++ b/cli/src/commands/ack.ts @@ -8,14 +8,15 @@ export function register(program: Command): void { program .command('ack ') .description('Acknowledge successful processing of a message (removes from inbox)') + .requiredOption('--lease-token ', 'Lease token returned by admp pull') .option('--result ', 'Optional result payload as JSON string') - .addHelpText('after', '\nExample:\n admp ack msg_abc123\n admp ack msg_abc123 --result \'{"status":"done"}\'') - .action(async (messageId: string, opts: { result?: string }) => { + .addHelpText('after', '\nExample:\n admp ack msg_abc123 --lease-token \n admp ack msg_abc123 --lease-token --result \'{"status":"done"}\'') + .action(async (messageId: string, opts: { result?: string; leaseToken: string }) => { validateMessageId(messageId); const config = requireConfig(['agent_id', 'secret_key', 'base_url']); const client = new AdmpClient(config); - const body: Record = {}; + const body: Record = { lease_token: opts.leaseToken }; if (opts.result) { try { body.result = JSON.parse(opts.result); diff --git a/cli/src/commands/nack.ts b/cli/src/commands/nack.ts index 90090ab..66072d5 100644 --- a/cli/src/commands/nack.ts +++ b/cli/src/commands/nack.ts @@ -8,22 +8,23 @@ export function register(program: Command): void { program .command('nack ') .description('Reject or defer a message (requeues for retry)') + .requiredOption('--lease-token ', 'Lease token returned by admp pull') .option('--extend ', 'Extend the lease by N seconds before requeuing') .option('--requeue', 'Force immediate requeue without waiting for lease to expire') - .addHelpText('after', '\nExample:\n admp nack msg_abc123\n admp nack msg_abc123 --extend 60 --requeue') - .action(async (messageId: string, opts: { extend?: string; requeue?: boolean }) => { + .addHelpText('after', '\nExample:\n admp nack msg_abc123 --lease-token \n admp nack msg_abc123 --lease-token --extend 60') + .action(async (messageId: string, opts: { extend?: string; requeue?: boolean; leaseToken: string }) => { validateMessageId(messageId); const config = requireConfig(['agent_id', 'secret_key', 'base_url']); const client = new AdmpClient(config); - const body: Record = {}; + const body: Record = { lease_token: opts.leaseToken }; if (opts.extend) { const n = parseInt(opts.extend, 10); if (isNaN(n) || n <= 0) { error(`--extend must be a positive integer, got: ${opts.extend}`, 'INVALID_ARGUMENT'); process.exit(1); } - body.extend = n; + body.extend_sec = n; } if (opts.requeue) body.requeue = true; diff --git a/cli/src/output.ts b/cli/src/output.ts index 1b7c513..0da3c22 100644 --- a/cli/src/output.ts +++ b/cli/src/output.ts @@ -61,7 +61,7 @@ export function printMessage(envelope: Record): void { return; } console.log(''); - console.log(bold('Message') + ' ' + dim(String(envelope.id ?? ''))); + console.log(bold('Message') + ' ' + dim(String(envelope.message_id ?? envelope.id ?? ''))); console.log(dim('─'.repeat(60))); console.log(` ${cyan('from')}: ${envelope.from ?? ''}`); console.log(` ${cyan('to')}: ${envelope.to ?? ''}`); @@ -73,11 +73,14 @@ export function printMessage(envelope: Record): void { if (envelope.lease_until) { console.log(` ${cyan('lease')}: ${envelope.lease_until}`); } + if (envelope.lease_token) { + console.log(` ${cyan('lease token')}: ${envelope.lease_token}`); + } if (envelope.attempts !== undefined) { console.log(` ${cyan('attempts')}: ${envelope.attempts}`); } console.log(''); console.log(bold('Body:')); - console.log(JSON.stringify(envelope.body, null, 2)); + console.log(JSON.stringify(envelope.envelope?.body ?? envelope.body, null, 2)); console.log(''); } diff --git a/docs/AGENT-GUIDE.md b/docs/AGENT-GUIDE.md index a27c96e..8cb6e1f 100644 --- a/docs/AGENT-GUIDE.md +++ b/docs/AGENT-GUIDE.md @@ -167,6 +167,7 @@ Response (200 OK with message, or 204 No Content if inbox is empty): "message_id": "uuid", "envelope": {...}, "lease_until": 1740000060000, + "lease_token": "opaque-per-claim-capability", "attempts": 1 } ``` @@ -175,14 +176,14 @@ Response (200 OK with message, or 204 No Content if inbox is empty): ```bash # CLI -admp ack +admp ack --lease-token # HTTP POST /api/agents/my-agent/messages//ack Signature: ... Date: ... -{"result": {"status": "processed"}} +{"result": {"status": "processed"}, "lease_token": ""} ``` --- @@ -562,7 +563,7 @@ Set `retain_until_acked: true` in the send body to require explicit acknowledgme Register with `auto_ack_on_pull: true` for fire-and-forget delivery. The hub immediately acks each message on pull — no explicit `POST .../ack` is required. -The pull response includes `"auto_acked": true` when the hub auto-acked the message. In this case `lease_until` is `null` — do not call `POST .../ack` for auto-acked messages (it will return 400). +The pull response includes `"auto_acked": true` when the hub auto-acked the message. In this case `lease_until` is `null` — do not call `POST .../ack` for auto-acked messages (it will return `409 LEASE_TOKEN_REQUIRED`). `retain_until_acked` always wins over `auto_ack_on_pull` — work orders and retained messages require explicit ack even if the recipient opted into auto-ack. diff --git a/docs/API-REFERENCE.md b/docs/API-REFERENCE.md index 0dddbcb..20dc0fe 100644 --- a/docs/API-REFERENCE.md +++ b/docs/API-REFERENCE.md @@ -526,6 +526,7 @@ Returns 204 (no content) when the inbox is empty. "timestamp": "2026-02-26T00:00:00Z" }, "lease_until": 1740000060000, + "lease_token": "opaque-per-claim-capability", "attempts": 1 } ``` @@ -536,7 +537,7 @@ Returns 204 (no content) when the inbox is empty. ### POST /api/agents/:agentId/messages/:messageId/ack -Acknowledge a message, confirming successful processing. The message must currently be in `leased` status. Ephemeral messages have their body purged on ack. +Acknowledge a message, confirming successful processing. The message must currently be in `leased` status and the `lease_token` must exactly match the token returned by its most recent pull. Ephemeral messages have their body purged on ack. **Auth:** HTTP Signature (must be the agent itself) @@ -545,6 +546,7 @@ Acknowledge a message, confirming successful processing. The message must curren | Field | Type | Required | Description | |-------|------|----------|-------------| | `result` | any | No | Processing result (stored with message record) | +| `lease_token` | string | Yes | Opaque capability returned by pull; required to fence stale placements | **Response 200:** ```json @@ -570,6 +572,9 @@ Negative acknowledge — either requeue the message or extend the current lease. |-------|------|----------|-------------| | `extend_sec` | number | No | Extend the lease by this many seconds from the current lease base | | `requeue` | boolean | No | Requeue immediately (default behavior if `extend_sec` not provided) | +| `lease_token` | string | Yes | Opaque capability returned by pull; required to fence stale placements | + +Both endpoints return `409 LEASE_TOKEN_REQUIRED` when the token is absent and `409 LEASE_FENCED` when the lease was reclaimed or the token is stale. **Response 200:** ```json diff --git a/docs/CLI-REFERENCE.md b/docs/CLI-REFERENCE.md index bc5905d..02974fd 100644 --- a/docs/CLI-REFERENCE.md +++ b/docs/CLI-REFERENCE.md @@ -69,8 +69,8 @@ ADMP_SEED=deadbeef... admp register --name my-agent |---------|-------------|-------| | `admp send` | Send a message to another agent's inbox. The envelope is signed with your Ed25519 key. Transport auth uses `api_key`. | `--to ` **(required)** Recipient agent ID. `--subject ` **(required)** Message type (e.g. `task.request`). `--body ` JSON body or `@filename` to read from file (relative paths only, max 1MB). Default: `{}`. `--type ` Message type field. Default: `task.request`. `--correlation-id ` Correlation ID for threading. `--ttl ` Time-to-live (max 86400). `--ephemeral` Do not persist message body after ack. `--json` | | `admp pull` | Pull the next message from your inbox. The message is leased (locked) until you ack or nack it. Returns empty message if inbox is empty. | `--timeout ` Long-poll timeout (max 300 seconds). Adds 5s buffer to client timeout to avoid racing the server. `--json` | -| `admp ack ` | Acknowledge a message, confirming successful processing. | `--result ` Optional JSON result to attach. `--json` | -| `admp nack ` | Reject or defer a message. | `--extend ` Extend the lease instead of requeuing. `--requeue` Explicitly requeue the message. `--json` | +| `admp ack ` | Acknowledge a message, confirming successful processing. | `--lease-token ` **(required)** Token returned by `admp pull`. `--result ` Optional JSON result to attach. `--json` | +| `admp nack ` | Reject or defer a message. | `--lease-token ` **(required)** Token returned by `admp pull`. `--extend ` Extend the lease instead of requeuing. `--requeue` Explicitly requeue the message. `--json` | | `admp reply ` | Send a correlated reply to a previously received message. The `correlation_id` is set automatically. | `--subject ` **(required)** Reply message type. `--body ` **(required)** JSON reply body. `--json` | | `admp status ` | Check the delivery status of a sent message. Returns lifecycle state (`queued`, `leased`, `acked`, `expired`, `purged`). | `--json` | | `admp inbox stats` | Show queue counts for your inbox. | `--json` | diff --git a/package.json b/package.json index 801a450..ba47b3f 100644 --- a/package.json +++ b/package.json @@ -19,6 +19,7 @@ "author": "Agent Dispatch Working Group", "license": "MIT", "dependencies": { + "@mech/storage-sdk": "0.3.2", "cors": "^2.8.5", "dotenv": "^17.2.3", "express": "^4.18.2", diff --git a/src/routes/inbox.js b/src/routes/inbox.js index 2623888..d46b2af 100644 --- a/src/routes/inbox.js +++ b/src/routes/inbox.js @@ -94,6 +94,7 @@ router.post('/:agentId/inbox/pull', authenticateHttpSignature, async (req, res) // auto_acked messages are already acked in storage; lease_until is not meaningful. lease_until: message.auto_acked ? null : message.lease_until, attempts: message.attempts, + ...(message.auto_acked ? {} : { lease_token: message.lease_token }), ...(message.auto_acked && { auto_acked: true }) }); } catch (error) { @@ -110,9 +111,9 @@ router.post('/:agentId/inbox/pull', authenticateHttpSignature, async (req, res) */ router.post('/:agentId/messages/:messageId/ack', authenticateHttpSignature, async (req, res) => { try { - const { result } = req.body; + const { result, lease_token } = req.body; - await inboxService.ack(req.params.agentId, req.params.messageId, result); + await inboxService.ack(req.params.agentId, req.params.messageId, result, lease_token); res.json({ ok: true }); } catch (error) { @@ -123,8 +124,8 @@ router.post('/:agentId/messages/:messageId/ack', authenticateHttpSignature, asyn }); } - res.status(400).json({ - error: 'ACK_FAILED', + res.status(error.statusCode || 400).json({ + error: error.code || 'ACK_FAILED', message: error.message }); } @@ -136,11 +137,12 @@ router.post('/:agentId/messages/:messageId/ack', authenticateHttpSignature, asyn */ router.post('/:agentId/messages/:messageId/nack', authenticateHttpSignature, async (req, res) => { try { - const { extend_sec, requeue } = req.body; + const { extend_sec, requeue, lease_token } = req.body; const message = await inboxService.nack(req.params.agentId, req.params.messageId, { extend_sec, - requeue + requeue, + lease_token }); res.json({ @@ -156,8 +158,8 @@ router.post('/:agentId/messages/:messageId/nack', authenticateHttpSignature, asy }); } - res.status(400).json({ - error: 'NACK_FAILED', + res.status(error.statusCode || 400).json({ + error: error.code || 'NACK_FAILED', message: error.message }); } diff --git a/src/server.test.js b/src/server.test.js index ccc76a7..302fc96 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -81,7 +81,9 @@ function withAgentHeader(req, agentId) { const MECH_CONFIGURED = !!createMechStorage && + process.env.RUN_MECH_STORAGE_INTEGRATION === '1' && process.env.STORAGE_BACKEND === 'mech' && + !!(process.env.MECH_BASE_URL || process.env.MECH_STORAGE_BASE_URL) && !!process.env.MECH_APP_ID && !!process.env.MECH_API_KEY; @@ -320,7 +322,8 @@ test('send → pull → ack → status flow', async () => { const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) .send({ - result: { status: 'success', note: 'ack from test' } + result: { status: 'success', note: 'ack from test' }, + lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); @@ -356,7 +359,7 @@ test('nack requeues message back to inbox', async () => { const nackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/nack`) - .send({ requeue: true }); + .send({ requeue: true, lease_token: firstPull.body.lease_token }); assert.equal(nackRes.status, 200); assert.equal(nackRes.body.ok, true); @@ -395,7 +398,7 @@ test('nack can extend lease without requeue', async () => { const nackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/nack`) - .send({ extend_sec: 30 }); + .send({ extend_sec: 30, lease_token: pullRes.body.lease_token }); assert.equal(nackRes.status, 200); assert.equal(nackRes.body.ok, true); @@ -443,6 +446,65 @@ test('reclaiming expired leases requeues messages', async () => { assert.ok(secondPull.body.envelope); }); +test('concurrent pulls conditionally claim a message only once', async () => { + const sender = await registerAgent('sender-conditional-claim'); + const recipient = await registerAgent('recipient-conditional-claim'); + const sendRes = await sendSignedMessage(sender, recipient.agent_id, { + subject: 'conditional-claim', + body: { test: 'conditional-claim' } + }); + assert.equal(sendRes.status, 201); + + const pullPath = `/api/agents/${encodeURIComponent(recipient.agent_id)}/inbox/pull`; + const [first, second] = await Promise.all([ + request(app).post(pullPath).send({ visibility_timeout: 60 }), + request(app).post(pullPath).send({ visibility_timeout: 60 }) + ]); + + const claimed = [first, second].filter(response => response.status === 200); + const empty = [first, second].filter(response => response.status === 204); + assert.equal(claimed.length, 1); + assert.equal(empty.length, 1); + assert.ok(claimed[0].body.lease_token); +}); + +test('stale or missing lease tokens cannot mutate a reclaimed message', async () => { + const sender = await registerAgent('sender-lease-fencing'); + const recipient = await registerAgent('recipient-lease-fencing'); + const sendRes = await sendSignedMessage(sender, recipient.agent_id, { + subject: 'lease-fencing', + body: { test: 'lease-fencing' } + }); + assert.equal(sendRes.status, 201); + const messageId = sendRes.body.message_id; + const pullPath = `/api/agents/${encodeURIComponent(recipient.agent_id)}/inbox/pull`; + const firstPull = await request(app).post(pullPath).send({ visibility_timeout: 60 }); + assert.equal(firstPull.status, 200); + + const missingToken = await request(app) + .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) + .send({}); + assert.equal(missingToken.status, 409); + assert.equal(missingToken.body.error, 'LEASE_TOKEN_REQUIRED'); + + await storage.updateMessage(messageId, { lease_until: Date.now() - 1 }); + await storage.expireLeases(); + const secondPull = await request(app).post(pullPath).send({ visibility_timeout: 60 }); + assert.equal(secondPull.status, 200); + assert.notEqual(secondPull.body.lease_token, firstPull.body.lease_token); + + const staleAck = await request(app) + .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) + .send({ lease_token: firstPull.body.lease_token }); + assert.equal(staleAck.status, 409); + assert.equal(staleAck.body.error, 'LEASE_FENCED'); + + const currentAck = await request(app) + .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) + .send({ lease_token: secondPull.body.lease_token }); + assert.equal(currentAck.status, 200); +}); + test('rejects messages with invalid signature', async () => { const sender = await registerAgent('sender-invalid-sig'); const recipient = await registerAgent('recipient-invalid-sig'); @@ -983,7 +1045,7 @@ test('ephemeral message: body purged on ack, metadata preserved', async () => { // Ack the message const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) - .send({ result: { status: 'processed' } }); + .send({ result: { status: 'processed' }, lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); @@ -1122,7 +1184,7 @@ test('non-ephemeral messages behave as before (backward compat)', async () => { const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) - .send({ result: { status: 'done' } }); + .send({ result: { status: 'done' }, lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); @@ -1168,7 +1230,7 @@ test('auto_ack_on_pull: message is immediately acked on pull, auto_acked returne const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) .send({}); - assert.equal(ackRes.status, 400); // already acked, not leased + assert.equal(ackRes.status, 409); // no current lease capability exists // Status confirms acked const statusRes = await request(app).get(`/api/messages/${messageId}/status`); @@ -1208,7 +1270,7 @@ test('retain_until_acked overrides auto_ack_on_pull — explicit ack required', // Explicit ack should succeed const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) - .send({}); + .send({ lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); assert.equal(ackRes.body.ok, true); }); @@ -1244,7 +1306,7 @@ test('work_order type always sets retain_until_acked server-side', async () => { // Explicit ack required const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) - .send({}); + .send({ lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); }); @@ -1277,7 +1339,7 @@ test('fix_request type always sets retain_until_acked server-side', async () => const ackRes = await request(app) .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) - .send({}); + .send({ lease_token: pullRes.body.lease_token }); assert.equal(ackRes.status, 200); }); diff --git a/src/services/inbox.service.js b/src/services/inbox.service.js index cac173a..79198b3 100644 --- a/src/services/inbox.service.js +++ b/src/services/inbox.service.js @@ -4,6 +4,7 @@ */ import { v4 as uuid } from 'uuid'; +import { randomUUID } from 'node:crypto'; import { storage } from '../storage/index.js'; import { verifySignature, fromBase64, validateTimestamp, parseTTL } from '../utils/crypto.js'; import { agentService } from './agent.service.js'; @@ -242,43 +243,34 @@ export class InboxService { autoAck = agent?.auto_ack_on_pull ?? false; } - // Get available messages - const now = Date.now(); - let messages = await storage.getInbox(agentId, 'queued'); - - // Filter out messages past their ephemeral TTL (security: don't serve expired secrets) - messages = messages.filter(m => !m.expires_at || m.expires_at > now); - - if (messages.length === 0) { - return null; - } + // Claiming must happen in storage, not after a read in this process: multiple + // hub instances may pull concurrently. The token is an opaque, per-claim + // capability that fences stale placements after a lease is reclaimed. + const leaseToken = randomUUID(); + const leaseUntil = Date.now() + (visibility_timeout * 1000); + const message = await storage.claimNextMessage(agentId, { + lease_until: leaseUntil, + lease_token: leaseToken + }); - // Get oldest message (FIFO) - const message = messages.sort((a, b) => a.created_at - b.created_at)[0]; + if (!message) return null; // auto_ack_on_pull: skip the lease and immediately ack. This avoids the state // inconsistency of returning a 'leased' snapshot while storage holds 'acked'. // If the ack write throws, it propagates — the message stays queued for retry. // retain_until_acked overrides auto_ack_on_pull regardless of agent preference. if (autoAck && !message.retain_until_acked) { - const acked = await storage.updateMessage(message.id, { + const acked = await storage.updateLeasedMessage(message.id, agentId, leaseToken, { status: 'acked', acked_at: Date.now(), - attempts: message.attempts + 1 + lease_until: null, + lease_token: null }); + if (!acked) throw this.leaseFencedError(); return { ...acked, auto_acked: true }; } - // Standard lease path - const leaseUntil = Date.now() + (visibility_timeout * 1000); - - const leased = await storage.updateMessage(message.id, { - status: 'leased', - lease_until: leaseUntil, - attempts: message.attempts + 1 - }); - - return leased; + return message; } /** @@ -288,7 +280,7 @@ export class InboxService { * @param {Object} result - Processing result * @returns {boolean} */ - async ack(agentId, messageId, result = {}) { + async ack(agentId, messageId, result = {}, leaseToken) { const message = await storage.getMessage(messageId); if (!message) { @@ -298,30 +290,33 @@ export class InboxService { if (message.to_agent_id !== agentId) { throw new Error('Message does not belong to this agent'); } - - if (message.status !== 'leased') { - throw new Error(`Message must be leased before ack (current status: ${message.status})`); - } + const currentLeaseToken = this.resolveLeaseToken(message, leaseToken); // If ephemeral, ack and purge body in a single update (no extra fetch) if (message.ephemeral) { const purgedEnvelope = { ...message.envelope }; delete purgedEnvelope.body; - await storage.updateMessage(messageId, { + const updated = await storage.updateLeasedMessage(messageId, agentId, currentLeaseToken, { status: 'purged', result, acked_at: Date.now(), envelope: purgedEnvelope, purged_at: Date.now(), - purge_reason: 'acked' + purge_reason: 'acked', + lease_until: null, + lease_token: null }); + if (!updated) throw this.leaseFencedError(); } else { - await storage.updateMessage(messageId, { + const updated = await storage.updateLeasedMessage(messageId, agentId, currentLeaseToken, { status: 'acked', result, - acked_at: Date.now() + acked_at: Date.now(), + lease_until: null, + lease_token: null }); + if (!updated) throw this.leaseFencedError(); } return true; @@ -347,22 +342,54 @@ export class InboxService { throw new Error('Message does not belong to this agent'); } + if (message.status !== 'leased') { + throw this.leaseFencedError(); + } + const currentLeaseToken = this.resolveLeaseToken(message, options.lease_token); + // Extend lease if (options.extend_sec) { const base = message.lease_until && message.lease_until > Date.now() ? message.lease_until : Date.now(); const newLeaseUntil = base + (options.extend_sec * 1000); - return await storage.updateMessage(messageId, { + const updated = await storage.updateLeasedMessage(messageId, agentId, currentLeaseToken, { lease_until: newLeaseUntil }); + if (!updated) throw this.leaseFencedError(); + return updated; } // Requeue - return await storage.updateMessage(messageId, { + const updated = await storage.updateLeasedMessage(messageId, agentId, currentLeaseToken, { status: 'queued', - lease_until: null + lease_until: null, + lease_token: null }); + if (!updated) throw this.leaseFencedError(); + return updated; + } + + resolveLeaseToken(message, leaseToken) { + if (typeof leaseToken === 'string' && leaseToken.length > 0 && leaseToken.length <= 255) { + return leaseToken; + } + if (process.env.LEASE_TOKEN_MODE === 'compat' && message?.lease_token) { + return message.lease_token; + } + { + const error = new Error('lease_token is required to mutate a leased message'); + error.code = 'LEASE_TOKEN_REQUIRED'; + error.statusCode = 409; + throw error; + } + } + + leaseFencedError() { + const error = new Error('Lease is no longer current'); + error.code = 'LEASE_FENCED'; + error.statusCode = 409; + return error; } /** diff --git a/src/storage/index.js b/src/storage/index.js index bc187fa..fad8e17 100644 --- a/src/storage/index.js +++ b/src/storage/index.js @@ -9,13 +9,8 @@ import { storage as memoryStorage } from './memory.js'; // Built-in backends: // memory — in-process Map-based storage (default, good for development) // -// Custom backends (overlay pattern): -// Place a compatible adapter at src/storage/.js and set STORAGE_BACKEND=. -// The adapter must export a `createMechStorage`-style factory or a singleton that -// implements the same interface as memory.js. See memory.js for the required methods. -// -// Example: STORAGE_BACKEND=mech loads ./mech.js (not shipped in the public repo; -// injected at deploy time via the agentdispatch-deploy overlay). +// Built-in persistent backend: +// mech — the supported SDK-backed adapter in ./mech.js. config(); @@ -25,18 +20,16 @@ let _storage; switch (backend) { case 'mech': { - // mech.js is a private adapter injected via the agentdispatch-deploy overlay. - // It is not included in the public repository. let mechMod; try { mechMod = await import('./mech.js'); } catch { throw new Error( - 'STORAGE_BACKEND=mech but src/storage/mech.js is not present. ' + - 'Provide the Mech Storage adapter via the agentdispatch-deploy overlay.' + 'STORAGE_BACKEND=mech but the bundled Mech Storage adapter could not be loaded.' ); } _storage = mechMod.createMechStorage(); + _storage.ensureConfigured(); break; } case 'memory': diff --git a/src/storage/mech.js b/src/storage/mech.js new file mode 100644 index 0000000..7ca63b0 --- /dev/null +++ b/src/storage/mech.js @@ -0,0 +1,1116 @@ +import { createStorageSdk } from '@mech/storage-sdk'; + +/** + * Mech Storage backend for ADMP + * Implements the same interface as MemoryStorage using Mech's NoSQL APIs. + */ + +export class MechStorage { + constructor({ baseUrl, appId, apiKey, sdk } = {}) { + this.baseUrl = baseUrl?.replace(/\/+$/, ''); + this.appId = appId; + this.apiKey = apiKey; + this.isInjectedSdk = Boolean(sdk); + this.sdk = sdk || (this.appId && this.apiKey + ? createStorageSdk({ baseUrl: this.baseUrl, apiKey: this.apiKey }) + : null); + } + + ensureConfigured() { + if (!this.appId || !this.apiKey || (!this.baseUrl && !this.isInjectedSdk)) { + throw new Error('Mech Storage is not configured. Set MECH_BASE_URL, MECH_APP_ID, and MECH_API_KEY.'); + } + } + + get nosql() { + this.ensureConfigured(); + return this.sdk.apps(this.appId).nosql; + } + + async createDocument(collection, documentKey, data) { + const result = await this.nosql.cas.createDocument({ + collection, + document_key: documentKey, + data + }); + if (!result.ok) { + const error = new Error(`Mech document already exists: ${collection}/${documentKey}`); + error.code = result.code; + throw error; + } + return result.document.data; + } + + async getDocument(collection, documentKey) { + const result = await this.nosql.cas.getDocument(collection, documentKey); + return result.ok && !result.document.data._deleted ? result.document : null; + } + + async getData(collection, documentKey) { + const document = await this.getDocument(collection, documentKey); + return document?.data || null; + } + + async listDocuments(collection) { + const documents = []; + const pageSize = 100; + let offset = 0; + let hasMore = true; + + while (hasMore) { + const result = await this.nosql.listDocuments({ + collection_name: collection, + limit: pageSize, + offset + }); + if (result.error) { + const error = new Error(result.error.message || `Mech document list failed for ${collection}`); + error.code = result.error.code; + throw error; + } + const page = result.data?.data || []; + documents.push(...page + .filter(document => !document.document?._deleted) + .map(document => document.document)); + hasMore = result.data?.pagination?.hasMore === true; + offset += page.length; + if (hasMore && page.length === 0) { + throw new Error(`Mech document pagination stalled for collection ${collection}`); + } + } + + return documents; + } + + async updateDocument(collection, documentKey, updates) { + // All updates are complete-record CAS writes. A conflict is retried from the + // returned current revision so unrelated concurrent fields are preserved. + let current = await this.getDocument(collection, documentKey); + for (let attempt = 0; current && attempt < 3; attempt++) { + const data = { ...current.data, ...updates }; + const result = await this.nosql.cas.updateDocument(collection, documentKey, { + _rev: current._rev, + data, + metadata: current.metadata + }); + if (result.ok) return result.document.data; + if (result.code === 'DOCUMENT_NOT_FOUND') return null; + current = result.current || await this.getDocument(collection, documentKey); + } + throw new Error(`Mech CAS update conflicted repeatedly: ${collection}/${documentKey}`); + } + + async tombstoneDocument(collection, documentKey) { + const updated = await this.updateDocument(collection, documentKey, { + _deleted: true, + deleted_at: Date.now(), + updated_at: Date.now() + }); + return updated !== null; + } + + async transitionDocument(collection, documentKey, transition, maxAttempts = 3) { + let current = await this.getDocument(collection, documentKey); + for (let attempt = 0; current && attempt < maxAttempts; attempt++) { + const next = transition(current.data); + if (!next) return null; + const result = await this.nosql.cas.updateDocument(collection, documentKey, { + _rev: current._rev, + data: next, + metadata: current.metadata + }); + if (result.ok) return result.document.data; + if (result.code === 'DOCUMENT_NOT_FOUND') return null; + current = result.current && !result.current.data._deleted + ? result.current + : await this.getDocument(collection, documentKey); + } + return null; + } + + async claimNextMessage(agentId, { lease_until, lease_token }) { + const now = Date.now(); + const candidates = (await this.listDocuments('admp_messages')) + .filter(message => message.to_agent_id === agentId + && message.status === 'queued' + && (!message.expires_at || message.expires_at > now)) + .sort((a, b) => a.created_at - b.created_at); + + for (const candidate of candidates) { + const claimed = await this.transitionDocument('admp_messages', candidate.id, current => { + if (current.to_agent_id !== agentId + || current.status !== 'queued' + || (current.expires_at && current.expires_at <= Date.now())) return null; + return { + ...current, + status: 'leased', + lease_until, + lease_token, + attempts: (current.attempts || 0) + 1, + updated_at: Date.now() + }; + }); + if (claimed) return claimed; + } + return null; + } + + async updateLeasedMessage(messageId, agentId, leaseToken, updates) { + return this.transitionDocument('admp_messages', messageId, current => { + if (current.to_agent_id !== agentId + || current.status !== 'leased' + || !leaseToken + || current.lease_token !== leaseToken + || !current.lease_until + || current.lease_until <= Date.now()) return null; + return { + ...current, + ...updates, + lease_token: updates.status && updates.status !== 'leased' ? null : current.lease_token, + updated_at: Date.now() + }; + }); + } + + extractDocument(wrapper) { + if (!wrapper) return null; + if (wrapper.document) return wrapper.document; + if (wrapper.data && wrapper.data.document) return wrapper.data.document; + return wrapper; + } + + extractDocuments(listJson) { + const docs = Array.isArray(listJson?.data) ? listJson.data : []; + return docs.map(doc => (doc.document ? doc.document : doc)); + } + + /** + * Compatibility bridge for the existing adapter methods. It intentionally + * does not perform HTTP itself: every operation is routed through the SDK's + * list/create/CAS primitives while the public storage interface is migrated. + */ + async request(path, { method = 'GET', body, allow404 = false } = {}) { + const keyLookup = path.match(/^\/nosql\/documents\/key\/([^?]+)\?collection_name=(.+)$/); + const documentPath = path.match(/^\/nosql\/documents\/([^/]+)\/([^/?]+)$/); + const legacyDeletePath = path.match(/^\/nosql\/documents\/([^?]+)\?collection_name=(.+)$/); + const listPath = path.match(/^\/nosql\/documents\?collection_name=([^&]+)/); + + if (method === 'POST' && path === '/nosql/documents') { + const data = await this.createDocument(body.collection_name, body.document_key, body.data); + return { status: 201, json: { data: { document: data } } }; + } + + if (method === 'GET' && keyLookup) { + const document = await this.getDocument(decodeURIComponent(keyLookup[2]), decodeURIComponent(keyLookup[1])); + if (!document || document.data._deleted) return { status: 404, json: null }; + return { status: 200, json: { data: { document: document.data } } }; + } + + if (method === 'GET' && listPath) { + const documents = await this.listDocuments(decodeURIComponent(listPath[1])); + return { status: 200, json: { data: documents.map(document => ({ document })) } }; + } + + if (method === 'PUT' && documentPath) { + const [collection, key] = [decodeURIComponent(documentPath[1]), decodeURIComponent(documentPath[2])]; + const data = await this.updateDocument(collection, key, body?.data || {}); + if (data) return { status: 200, json: { data: { document: data } } }; + const error = new Error(`Mech document not found: ${collection}/${key}`); + error.status = 404; + throw error; + } + + if (method === 'DELETE' && (documentPath || legacyDeletePath)) { + const collection = decodeURIComponent(documentPath ? documentPath[1] : legacyDeletePath[2]); + const key = decodeURIComponent(documentPath ? documentPath[2] : legacyDeletePath[1]); + // The SDK has no physical delete primitive. Preserve auditability until + // retention is explicitly designed by marking the document deleted. + const removed = await this.tombstoneDocument(collection, key); + return { status: removed ? 200 : 404, json: null }; + } + + const error = new Error(`Unsupported Mech SDK operation: ${method} ${path}`); + error.status = 400; + if (allow404) return { status: 404, json: null }; + throw error; + } + + domainConfigKey(agentId) { + return `domain:${agentId}`; + } + + // ============ AGENTS ============ + + async createAgent(agent) { + const now = Date.now(); + const stored = { + ...agent, + created_at: now, + updated_at: now + }; + + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_agents', + document_key: stored.agent_id, + data: stored + } + }); + + return stored; + } + + async getAgent(agentId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(agentId)}?collection_name=admp_agents`, + { allow404: true } + ); + + if (status === 404) { + return null; + } + + const agentDoc = json?.data; + const agent = this.extractDocument(agentDoc); + return agent || null; + } + + async updateAgent(agentId, updates) { + const now = Date.now(); + const patch = { + ...updates, + updated_at: now + }; + + await this.request(`/nosql/documents/admp_agents/${encodeURIComponent(agentId)}`, { + method: 'PUT', + body: { + data: patch + } + }); + + return this.getAgent(agentId); + } + + async deleteAgent(agentId) { + const { status } = await this.request( + `/nosql/documents/admp_agents/${encodeURIComponent(agentId)}`, + { method: 'DELETE', allow404: true } + ); + + return status === 200 || status === 204; + } + + async listAgents(filter = {}) { + const { json } = await this.request('/nosql/documents?collection_name=admp_agents&limit=1000'); + let agents = this.extractDocuments(json); + + if (filter.status) { + agents = agents.filter(a => a.heartbeat?.status === filter.status); + } + + if (filter.registration_status) { + agents = agents.filter(a => a.registration_status === filter.registration_status); + } + + if (filter.tenant_id) { + agents = agents.filter(a => a.tenant_id === filter.tenant_id); + } + + return agents; + } + + async getAgentByDid(did) { + const { json } = await this.request('/nosql/documents?collection_name=admp_agents&limit=1000'); + const agents = this.extractDocuments(json); + return agents.find(a => a.did === did) || null; + } + + // ============ TENANTS ============ + + async createTenant(tenant) { + const now = Date.now(); + const stored = { + ...tenant, + created_at: now, + updated_at: now + }; + + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_tenants', + document_key: stored.tenant_id, + data: stored + } + }); + + return stored; + } + + async getTenant(tenantId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(tenantId)}?collection_name=admp_tenants`, + { allow404: true } + ); + + if (status === 404) return null; + + const doc = json?.data; + return this.extractDocument(doc) || null; + } + + async updateTenant(tenantId, updates) { + const now = Date.now(); + const patch = { + ...updates, + updated_at: now + }; + + await this.request(`/nosql/documents/admp_tenants/${encodeURIComponent(tenantId)}`, { + method: 'PUT', + body: { data: patch } + }); + + return this.getTenant(tenantId); + } + + async deleteTenant(tenantId) { + const { status } = await this.request( + `/nosql/documents/admp_tenants/${encodeURIComponent(tenantId)}`, + { method: 'DELETE', allow404: true } + ); + return status === 200 || status === 204; + } + + async listTenants() { + const { json } = await this.request('/nosql/documents?collection_name=admp_tenants&limit=1000'); + return this.extractDocuments(json); + } + + async getAgentsByTenant(tenantId) { + const { json } = await this.request('/nosql/documents?collection_name=admp_agents&limit=1000'); + const agents = this.extractDocuments(json); + return agents.filter(a => a.tenant_id === tenantId); + } + + // ============ MESSAGES / INBOX ============ + + async createMessage(message) { + const now = Date.now(); + const stored = { + ...message, + created_at: now, + updated_at: now + }; + + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_messages', + document_key: stored.id, + data: stored + } + }); + + return stored; + } + + async getMessage(messageId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(messageId)}?collection_name=admp_messages`, + { allow404: true } + ); + + if (status === 404) { + return null; + } + + const msgDoc = json?.data; + const message = this.extractDocument(msgDoc); + return message || null; + } + + async updateMessage(messageId, updates) { + const now = Date.now(); + const patch = { + ...updates, + updated_at: now + }; + + await this.request(`/nosql/documents/admp_messages/${encodeURIComponent(messageId)}`, { + method: 'PUT', + body: { + data: patch + } + }); + + return this.getMessage(messageId); + } + + async deleteMessage(messageId) { + const { status } = await this.request( + `/nosql/documents/admp_messages/${encodeURIComponent(messageId)}`, + { method: 'DELETE', allow404: true } + ); + + return status === 200 || status === 204; + } + + async getInbox(agentId, status = null) { + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + let messages = this.extractDocuments(json).filter(m => m.to_agent_id === agentId); + + if (status) { + messages = messages.filter(m => m.status === status); + } + + return messages; + } + + async getInboxStats(agentId) { + const messages = await this.getInbox(agentId); + + return { + total: messages.length, + queued: messages.filter(m => m.status === 'queued').length, + leased: messages.filter(m => m.status === 'leased').length, + acked: messages.filter(m => m.status === 'acked').length, + failed: messages.filter(m => m.status === 'failed').length + }; + } + + // ============ CLEANUP / STATS ============ + + async expireLeases() { + const now = Date.now(); + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + const messages = this.extractDocuments(json); + + let expired = 0; + + for (const message of messages) { + if (message.status === 'leased' && message.lease_until && message.lease_until < now) { + const requeued = await this.transitionDocument('admp_messages', message.id, current => { + if (current.status !== 'leased' || !current.lease_until || current.lease_until >= Date.now()) return null; + return { ...current, status: 'queued', lease_until: null, lease_token: null, updated_at: Date.now() }; + }); + if (requeued) expired++; + } + } + + return expired; + } + + async expireMessages() { + const now = Date.now(); + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + const messages = this.extractDocuments(json); + + let expired = 0; + + for (const message of messages) { + const age = now - message.created_at; + const ttl = message.ttl_sec * 1000; + + if (age > ttl) { + await this.updateMessage(message.id, { + status: 'expired' + }); + expired++; + } + } + + return expired; + } + + async cleanupExpiredMessages() { + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + const messages = this.extractDocuments(json); + + let deleted = 0; + + for (const message of messages) { + if (message.status === 'expired' || message.status === 'acked') { + const age = Date.now() - message.updated_at; + if (age > 3600000) { + const removed = await this.deleteMessage(message.id); + if (removed) { + deleted++; + } + } + } + } + + return deleted; + } + + // TODO: The limit=1000 cap applies to all message list operations (purge, expire, + // cleanup). Messages beyond the 1000th won't be processed in a single sweep. + // For high-volume deployments, implement pagination or storage-side filtering. + async purgeExpiredEphemeralMessages() { + const now = Date.now(); + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + const messages = this.extractDocuments(json); + + let purged = 0; + + for (const message of messages) { + if (message.expires_at && message.expires_at < now && message.status !== 'purged') { + const purgedEnvelope = { ...message.envelope }; + delete purgedEnvelope.body; + + await this.updateMessage(message.id, { + status: 'purged', + envelope: purgedEnvelope, + purged_at: now, + purge_reason: 'ttl_expired' + }); + purged++; + } + } + + return purged; + } + + async getStats() { + const { json: agentsJson } = await this.request('/nosql/documents?collection_name=admp_agents&limit=1000'); + const { json: messagesJson } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + const { json: groupsJson } = await this.request('/nosql/documents?collection_name=admp_groups&limit=1000'); + + const agents = this.extractDocuments(agentsJson); + const messages = this.extractDocuments(messagesJson); + const groups = this.extractDocuments(groupsJson); + + return { + agents: { + total: agents.length, + online: agents.filter(a => a.heartbeat?.status === 'online').length, + offline: agents.filter(a => a.heartbeat?.status === 'offline').length + }, + messages: { + total: messages.length, + queued: messages.filter(m => m.status === 'queued').length, + leased: messages.filter(m => m.status === 'leased').length, + acked: messages.filter(m => m.status === 'acked').length, + failed: messages.filter(m => m.status === 'failed').length, + expired: messages.filter(m => m.status === 'expired').length, + purged: messages.filter(m => m.status === 'purged').length + }, + groups: { + total: groups.length + } + }; + } + + // ============ GROUPS ============ + + async createGroup(group) { + const now = Date.now(); + const stored = { + ...group, + created_at: now, + updated_at: now + }; + + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_groups', + document_key: stored.id, + data: stored + } + }); + + return stored; + } + + async getGroup(groupId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(groupId)}?collection_name=admp_groups`, + { allow404: true } + ); + + if (status === 404) { + return null; + } + + const groupDoc = json?.data; + const group = this.extractDocument(groupDoc); + return group || null; + } + + async updateGroup(groupId, updates) { + const now = Date.now(); + const patch = { + ...updates, + updated_at: now + }; + + await this.request(`/nosql/documents/admp_groups/${encodeURIComponent(groupId)}`, { + method: 'PUT', + body: { + data: patch + } + }); + + return this.getGroup(groupId); + } + + async deleteGroup(groupId) { + const { status } = await this.request( + `/nosql/documents/admp_groups/${encodeURIComponent(groupId)}`, + { method: 'DELETE', allow404: true } + ); + + return status === 200 || status === 204; + } + + async listGroups(filter = {}) { + const { json } = await this.request('/nosql/documents?collection_name=admp_groups&limit=1000'); + let groups = this.extractDocuments(json); + + if (filter.member) { + groups = groups.filter(g => g.members?.some(m => m.agent_id === filter.member)); + } + + return groups; + } + + // ============ GROUP MEMBERS ============ + + async addGroupMember(groupId, member) { + const group = await this.getGroup(groupId); + if (!group) { + throw new Error(`Group ${groupId} not found`); + } + + const members = group.members || []; + + // Check if already a member + if (members.some(m => m.agent_id === member.agent_id)) { + throw new Error(`Agent ${member.agent_id} is already a member`); + } + + const newMember = { + ...member, + joined_at: Date.now() + }; + + members.push(newMember); + + return this.updateGroup(groupId, { members }); + } + + async removeGroupMember(groupId, agentId) { + const group = await this.getGroup(groupId); + if (!group) { + throw new Error(`Group ${groupId} not found`); + } + + const members = (group.members || []).filter(m => m.agent_id !== agentId); + + return this.updateGroup(groupId, { members }); + } + + async getGroupMembers(groupId) { + const group = await this.getGroup(groupId); + if (!group) { + throw new Error(`Group ${groupId} not found`); + } + + return group.members || []; + } + + async isGroupMember(groupId, agentId) { + const group = await this.getGroup(groupId); + if (!group) { + return false; + } + + return (group.members || []).some(m => m.agent_id === agentId); + } + + // ============ GROUP MESSAGES ============ + + async getGroupMessages(groupId, options = {}) { + const { json } = await this.request('/nosql/documents?collection_name=admp_messages&limit=1000'); + // group_id is stored in the envelope, not at top level + let messages = this.extractDocuments(json).filter(m => + m.group_id === groupId || m.envelope?.group_id === groupId + ); + + // Deduplicate by group_message_id (each group post is fanned out to multiple recipients) + const seen = new Set(); + messages = messages.filter(m => { + const groupMsgId = m.envelope?.group_message_id || m.group_message_id || m.id; + if (seen.has(groupMsgId)) { + return false; + } + seen.add(groupMsgId); + return true; + }); + + // Sort by timestamp descending (newest first) + messages.sort((a, b) => b.created_at - a.created_at); + + // Apply limit + if (options.limit) { + messages = messages.slice(0, options.limit); + } + + // Return envelope data for history view + return messages.map(m => ({ + id: m.envelope?.group_message_id || m.group_message_id || m.id, + from: m.from_agent_id, + subject: m.envelope?.subject, + body: m.envelope?.body, + timestamp: m.envelope?.timestamp || m.created_at, + group_id: m.envelope?.group_id || m.group_id + })); + } + + // ============ DOMAINS ============ + + async setDomainConfig(agentId, config) { + const now = Date.now(); + const existing = await this.getDomainConfig(agentId); + const docKey = this.domainConfigKey(agentId); + + const stored = { + ...config, + agent_id: agentId, + created_at: existing?.created_at || now, + updated_at: now + }; + + try { + await this.request(`/nosql/documents/admp_domains/${encodeURIComponent(docKey)}`, { + method: 'PUT', + body: { data: stored } + }); + } catch (error) { + if (error.status !== 404) throw error; + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_domains', + document_key: docKey, + data: stored + } + }); + } + + return stored; + } + + async getDomainConfig(agentId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(this.domainConfigKey(agentId))}?collection_name=admp_domains`, + { allow404: true } + ); + + if (status !== 404) { + const doc = this.extractDocument(json?.data); + if (doc?.agent_id === agentId && typeof doc?.domain === 'string') { + return doc; + } + } + + // Backward-compat: legacy records keyed by raw agentId. + const legacy = await this.request( + `/nosql/documents/key/${encodeURIComponent(agentId)}?collection_name=admp_domains`, + { allow404: true } + ); + if (legacy.status === 404) return null; + + const legacyDoc = this.extractDocument(legacy.json?.data); + if (legacyDoc?.agent_id === agentId && typeof legacyDoc?.domain === 'string') { + return legacyDoc; + } + + return null; + } + + async deleteDomainConfig(agentId) { + const primary = await this.request( + `/nosql/documents/admp_domains/${encodeURIComponent(this.domainConfigKey(agentId))}`, + { method: 'DELETE', allow404: true } + ); + const legacy = await this.request( + `/nosql/documents/admp_domains/${encodeURIComponent(agentId)}`, + { method: 'DELETE', allow404: true } + ); + return ( + primary.status === 200 || primary.status === 204 || + legacy.status === 200 || legacy.status === 204 + ); + } + + // ============ ISSUED API KEYS ============ + + async createIssuedKey(key) { + const stored = { ...key, created_at: key.created_at || Date.now() }; + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_api_keys', + document_key: stored.key_id, + data: stored + } + }); + // Write a hash-indexed pointer document for O(1) getIssuedKeyByHash lookups. + // The pointer only stores the key_id; revocation/expiry is checked on the primary record. + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_api_key_hashes', + document_key: stored.key_hash, + data: { key_id: stored.key_id } + } + }); + return stored; + } + + async getIssuedKey(keyId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(keyId)}?collection_name=admp_api_keys`, + { allow404: true } + ); + if (status === 404) return null; + return this.extractDocument(json?.data) || null; + } + + async getIssuedKeyByHash(keyHash) { + // O(1) lookup via hash-index collection (written in createIssuedKey). + // Falls back to full scan if the index doesn't exist (e.g. keys created before this change). + const { status: idxStatus, json: idxJson } = await this.request( + `/nosql/documents/key/${encodeURIComponent(keyHash)}?collection_name=admp_api_key_hashes`, + { allow404: true } + ); + if (idxStatus !== 404) { + const pointer = this.extractDocument(idxJson?.data); + if (pointer?.key_id) { + return this.getIssuedKey(pointer.key_id); + } + } + // Fallback: linear scan (catches pre-index keys; remove once all keys are re-issued) + // If this warning fires, re-issue API keys so the hash index gets populated. + console.warn('[mech] admp_api_key_hashes index miss — falling back to linear scan. Re-issue API keys to rebuild the index.'); + const { json } = await this.request('/nosql/documents?collection_name=admp_api_keys&limit=1000'); + const keys = this.extractDocuments(json); + return keys.find(k => k.key_hash === keyHash) || null; + } + + async listIssuedKeys() { + const { json } = await this.request('/nosql/documents?collection_name=admp_api_keys&limit=1000'); + return this.extractDocuments(json); + } + + async revokeIssuedKey(keyId) { + const key = await this.getIssuedKey(keyId); + if (!key) return false; + // Mech API uses different URL forms per operation: + // GET → /nosql/documents/key/:key?collection_name=... (query-param format) + // PUT → /nosql/documents/:collection/:key (path-segment format) + // this.request() throws on non-2xx, so reaching `return true` implies success. + await this.request(`/nosql/documents/admp_api_keys/${encodeURIComponent(keyId)}`, { + method: 'PUT', + body: { data: { ...key, revoked: true, revoked_at: Date.now() } } + }); + return true; + } + + async updateIssuedKey(keyId, updates) { + const key = await this.getIssuedKey(keyId); + if (!key) return null; + const updated = { ...key, ...updates }; + // See revokeIssuedKey for note on Mech URL format difference between GET and PUT. + await this.request(`/nosql/documents/admp_api_keys/${encodeURIComponent(keyId)}`, { + method: 'PUT', + body: { data: updated } + }); + return updated; + } + + /** + * Atomically burn a single-use token: sets used_at only if it is currently null. + * Returns true if this call burned the token, false if it was already burned. + * + * NOTE: Mech backend does not support conditional writes natively, so this uses + * read-then-conditional-write. The race window is narrower than the old + * unconditional write (we reject if used_at is already set after re-read), + * but is not fully atomic. For true atomicity, migrate to a backend that + * supports conditional updates (e.g. PostgreSQL WHERE used_at IS NULL). + */ + async burnSingleUseKey(keyId) { + const key = await this.getIssuedKey(keyId); + if (!key || key.used_at) return false; + const updated = { ...key, used_at: Date.now() }; + await this.request(`/nosql/documents/admp_api_keys/${encodeURIComponent(keyId)}`, { + method: 'PUT', + body: { data: updated } + }); + return true; + } + + // ============ OUTBOX ============ + + async createOutboxMessage(message) { + const now = Date.now(); + const stored = { + ...message, + created_at: now, + updated_at: now + }; + + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_outbox', + document_key: stored.id, + data: stored + } + }); + + return stored; + } + + async getOutboxMessage(messageId) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(messageId)}?collection_name=admp_outbox`, + { allow404: true } + ); + + if (status === 404) return null; + + const doc = json?.data; + return this.extractDocument(doc) || null; + } + + async updateOutboxMessage(messageId, updates) { + // Fetch-then-merge to avoid losing fields on PUT (Mech replaces the document) + const existing = await this.getOutboxMessage(messageId); + if (!existing) return null; + + const merged = { + ...existing, + ...updates, + updated_at: Date.now() + }; + + await this.request(`/nosql/documents/admp_outbox/${encodeURIComponent(messageId)}`, { + method: 'PUT', + body: { data: merged } + }); + + return merged; + } + + async findOutboxMessageByProviderId(providerId) { + const { json } = await this.request('/nosql/documents?collection_name=admp_outbox&limit=1000'); + const messages = this.extractDocuments(json); + + for (const msg of messages) { + if (msg.provider_message_id === providerId) { + return msg; + } + } + + return null; + } + + async getOutboxMessages(agentId, options = {}) { + const { json } = await this.request('/nosql/documents?collection_name=admp_outbox&limit=1000'); + let messages = this.extractDocuments(json).filter(m => m.agent_id === agentId); + + if (options.status) { + messages = messages.filter(m => m.status === options.status); + } + + messages.sort((a, b) => b.created_at - a.created_at); + + if (options.limit) { + messages = messages.slice(0, options.limit); + } + + return messages; + } + + // ============ ROUND TABLES ============ + + async createRoundTable(rt) { + await this.request('/nosql/documents', { + method: 'POST', + body: { + collection_name: 'admp_round_tables', + document_key: rt.id, + data: rt + } + }); + return rt; + } + + async getRoundTable(id) { + const { status, json } = await this.request( + `/nosql/documents/key/${encodeURIComponent(id)}?collection_name=admp_round_tables`, + { allow404: true } + ); + if (status === 404) return null; + return this.extractDocument(json?.data) || null; + } + + async updateRoundTable(id, updates) { + const existing = await this.getRoundTable(id); + if (!existing) return null; + const updated = { ...existing, ...updates, updated_at: new Date().toISOString() }; + await this.request(`/nosql/documents/admp_round_tables/${encodeURIComponent(id)}`, { + method: 'PUT', + body: { data: updated } + }); + return updated; + } + + async listRoundTables(filter = {}) { + const { json } = await this.request('/nosql/documents?collection_name=admp_round_tables&limit=1000'); + let tables = this.extractDocuments(json); + if (filter.status) { + tables = tables.filter(rt => rt.status === filter.status); + } + if (filter.participant) { + tables = tables.filter(rt => + rt.facilitator === filter.participant || + (rt.participants || []).includes(filter.participant) + ); + } + return tables; + } + + async purgeStaleRoundTables(olderThanMs) { + const { json } = await this.request('/nosql/documents?collection_name=admp_round_tables&limit=1000'); + const tables = this.extractDocuments(json); + const cutoff = Date.now() - olderThanMs; + let purged = 0; + for (const rt of tables) { + if (rt.status === 'resolved' || rt.status === 'expired') { + const closedAt = rt.resolved_at || rt.expires_at; + if (closedAt && new Date(closedAt).getTime() < cutoff) { + try { + await this.request(`/nosql/documents/${rt.id}?collection_name=admp_round_tables`, { method: 'DELETE' }); + purged++; + } catch (_) {} + } + } + } + return purged; + } +} + +export function createMechStorage() { + return new MechStorage({ + baseUrl: process.env.MECH_BASE_URL || process.env.MECH_STORAGE_BASE_URL, + appId: process.env.MECH_APP_ID || process.env.MECH_STORAGE_APP_ID, + apiKey: process.env.MECH_API_KEY || process.env.MECH_STORAGE_API_KEY + }); +} diff --git a/src/storage/mech.test.js b/src/storage/mech.test.js new file mode 100644 index 0000000..05b247a --- /dev/null +++ b/src/storage/mech.test.js @@ -0,0 +1,93 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { MechStorage } from './mech.js'; + +function fakeSdk() { + const documents = new Map(); + const keyFor = (collection, key) => `${collection}:${key}`; + const cas = { + async getDocument(collection, key) { + const document = documents.get(keyFor(collection, key)); + return document + ? { ok: true, document: structuredClone(document) } + : { ok: false, code: 'DOCUMENT_NOT_FOUND' }; + }, + async createDocument({ collection, document_key, data }) { + const id = keyFor(collection, document_key); + if (documents.has(id)) return { ok: false, code: 'DOCUMENT_EXISTS' }; + const document = { collection, document_key, data: structuredClone(data), metadata: {}, _rev: '1', id, created_at: '', updated_at: '' }; + documents.set(id, document); + return { ok: true, document: structuredClone(document) }; + }, + async updateDocument(collection, key, { _rev, data, metadata }) { + const id = keyFor(collection, key); + const current = documents.get(id); + if (!current) return { ok: false, code: 'DOCUMENT_NOT_FOUND' }; + if (current._rev !== _rev) return { ok: false, code: 'REVISION_CONFLICT', current: structuredClone(current) }; + const document = { ...current, data: structuredClone(data), metadata, _rev: String(Number(_rev) + 1) }; + documents.set(id, document); + return { ok: true, document: structuredClone(document) }; + } + }; + return { + apps() { + return { + nosql: { + cas, + async listDocuments({ collection_name, limit = 100, offset = 0 }) { + const all = [...documents.values()] + .filter(d => d.collection === collection_name) + .map(document => ({ document: structuredClone(document.data) })); + const page = all.slice(offset, offset + limit); + return { data: { + data: page, + pagination: { hasMore: offset + page.length < all.length } + } }; + } + } + }; + } + }; +} + +test('MechStorage merges full records through CAS and tombstones instead of deleting', async () => { + const storage = new MechStorage({ appId: 'test', apiKey: 'test', sdk: fakeSdk() }); + await storage.createAgent({ agent_id: 'agent-1', did: 'did:key:one', preserved: true }); + + const updated = await storage.updateAgent('agent-1', { heartbeat: { status: 'online' } }); + assert.equal(updated.preserved, true); + assert.deepEqual(updated.heartbeat, { status: 'online' }); + + assert.equal(await storage.deleteAgent('agent-1'), true); + assert.equal(await storage.getAgent('agent-1'), null); + assert.deepEqual(await storage.listAgents(), []); +}); + +test('MechStorage pages collections and fences concurrent message claims', async () => { + const storage = new MechStorage({ appId: 'test', apiKey: 'test', sdk: fakeSdk() }); + const now = Date.now(); + for (let i = 0; i < 101; i++) { + await storage.createAgent({ agent_id: `agent-${i}`, did: `did:key:${i}` }); + } + assert.equal((await storage.listAgents()).length, 101); + + await storage.createMessage({ + id: 'message-1', + to_agent_id: 'recipient', + status: 'queued', + attempts: 0, + created_at: now, + envelope: { subject: 'test' } + }); + + const [first, second] = await Promise.all([ + storage.claimNextMessage('recipient', { lease_until: now + 60_000, lease_token: 'token-a' }), + storage.claimNextMessage('recipient', { lease_until: now + 60_000, lease_token: 'token-b' }) + ]); + const claimed = [first, second].filter(Boolean); + assert.equal(claimed.length, 1); + const winningToken = claimed[0].lease_token; + const losingToken = winningToken === 'token-a' ? 'token-b' : 'token-a'; + assert.equal(await storage.updateLeasedMessage('message-1', 'recipient', losingToken, { status: 'acked' }), null); + assert.equal((await storage.updateLeasedMessage('message-1', 'recipient', winningToken, { status: 'acked' })).status, 'acked'); +}); diff --git a/src/storage/memory.js b/src/storage/memory.js index 40cb86d..84a3628 100644 --- a/src/storage/memory.js +++ b/src/storage/memory.js @@ -157,6 +157,62 @@ export class MemoryStorage { return updated; } + /** + * Atomically lease the next eligible queued message for an agent. + * + * The production adapter must provide the same compare-and-set guarantee. In + * memory, the selection and mutation happen without an await boundary, so two + * concurrent callers cannot both claim the same record. + */ + async claimNextMessage(agentId, { lease_until, lease_token }) { + const now = Date.now(); + const messageIds = this.inboxes.get(agentId) || []; + const message = messageIds + .map(id => this.messages.get(id)) + .filter(message => message + && message.status === 'queued' + && (!message.expires_at || message.expires_at > now)) + .sort((a, b) => a.created_at - b.created_at)[0]; + + if (!message) return null; + + const claimed = { + ...message, + status: 'leased', + lease_until, + lease_token, + attempts: (message.attempts || 0) + 1, + updated_at: now + }; + this.messages.set(message.id, claimed); + return claimed; + } + + /** + * Atomically mutate a currently leased message only when the lease capability + * still matches. Returning null deliberately does not reveal whether a token + * was stale, incorrect, or the lease was reclaimed. + */ + async updateLeasedMessage(messageId, agentId, leaseToken, updates) { + const message = this.messages.get(messageId); + if (!message + || message.to_agent_id !== agentId + || message.status !== 'leased' + || message.lease_token !== leaseToken + || !message.lease_until + || message.lease_until <= Date.now()) { + return null; + } + + const updated = { + ...message, + ...updates, + updated_at: Date.now() + }; + this.messages.set(messageId, updated); + return updated; + } + async deleteMessage(messageId) { const message = this.messages.get(messageId); if (!message) return false; @@ -203,11 +259,17 @@ export class MemoryStorage { for (const message of this.messages.values()) { if (message.status === 'leased' && message.lease_until && message.lease_until < now) { - await this.updateMessage(message.id, { - status: 'queued', - lease_until: null - }); - expired++; + const latest = this.messages.get(message.id); + if (latest?.status === 'leased' && latest.lease_until && latest.lease_until < Date.now()) { + this.messages.set(message.id, { + ...latest, + status: 'queued', + lease_until: null, + lease_token: null, + updated_at: Date.now() + }); + expired++; + } } } From c1b6c7fb68df408c5b2988a36243641dc4740d83 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:26:16 -0500 Subject: [PATCH 05/10] test(inbox): reject expired lease mutations Covers stale lease-token acknowledgement before requeue. --- src/server.test.js | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/server.test.js b/src/server.test.js index 302fc96..6ee2098 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -488,6 +488,11 @@ test('stale or missing lease tokens cannot mutate a reclaimed message', async () assert.equal(missingToken.body.error, 'LEASE_TOKEN_REQUIRED'); await storage.updateMessage(messageId, { lease_until: Date.now() - 1 }); + const expiredAck = await request(app) + .post(`/api/agents/${encodeURIComponent(recipient.agent_id)}/messages/${messageId}/ack`) + .send({ lease_token: firstPull.body.lease_token }); + assert.equal(expiredAck.status, 409); + assert.equal(expiredAck.body.error, 'LEASE_FENCED'); await storage.expireLeases(); const secondPull = await request(app).post(pullPath).send({ visibility_timeout: 60 }); assert.equal(secondPull.status, 200); From ab3f46a9ae81f894e52c04303b1aa622b1b741ca Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:26:57 -0500 Subject: [PATCH 06/10] fix(storage): make issued key consumption conditional Use CAS for single-use key burn and paginate key lookups. --- src/storage/mech.js | 28 +++++++--------------------- src/storage/mech.test.js | 11 +++++++++++ 2 files changed, 18 insertions(+), 21 deletions(-) diff --git a/src/storage/mech.js b/src/storage/mech.js index 7ca63b0..4357b52 100644 --- a/src/storage/mech.js +++ b/src/storage/mech.js @@ -898,14 +898,12 @@ export class MechStorage { // Fallback: linear scan (catches pre-index keys; remove once all keys are re-issued) // If this warning fires, re-issue API keys so the hash index gets populated. console.warn('[mech] admp_api_key_hashes index miss — falling back to linear scan. Re-issue API keys to rebuild the index.'); - const { json } = await this.request('/nosql/documents?collection_name=admp_api_keys&limit=1000'); - const keys = this.extractDocuments(json); + const keys = await this.listDocuments('admp_api_keys'); return keys.find(k => k.key_hash === keyHash) || null; } async listIssuedKeys() { - const { json } = await this.request('/nosql/documents?collection_name=admp_api_keys&limit=1000'); - return this.extractDocuments(json); + return this.listDocuments('admp_api_keys'); } async revokeIssuedKey(keyId) { @@ -934,25 +932,13 @@ export class MechStorage { return updated; } - /** - * Atomically burn a single-use token: sets used_at only if it is currently null. - * Returns true if this call burned the token, false if it was already burned. - * - * NOTE: Mech backend does not support conditional writes natively, so this uses - * read-then-conditional-write. The race window is narrower than the old - * unconditional write (we reject if used_at is already set after re-read), - * but is not fully atomic. For true atomicity, migrate to a backend that - * supports conditional updates (e.g. PostgreSQL WHERE used_at IS NULL). - */ + /** Atomically burn a single-use token only if it has not already been used. */ async burnSingleUseKey(keyId) { - const key = await this.getIssuedKey(keyId); - if (!key || key.used_at) return false; - const updated = { ...key, used_at: Date.now() }; - await this.request(`/nosql/documents/admp_api_keys/${encodeURIComponent(keyId)}`, { - method: 'PUT', - body: { data: updated } + const burned = await this.transitionDocument('admp_api_keys', keyId, current => { + if (current.used_at) return null; + return { ...current, used_at: Date.now(), updated_at: Date.now() }; }); - return true; + return Boolean(burned); } // ============ OUTBOX ============ diff --git a/src/storage/mech.test.js b/src/storage/mech.test.js index 05b247a..225e496 100644 --- a/src/storage/mech.test.js +++ b/src/storage/mech.test.js @@ -91,3 +91,14 @@ test('MechStorage pages collections and fences concurrent message claims', async assert.equal(await storage.updateLeasedMessage('message-1', 'recipient', losingToken, { status: 'acked' }), null); assert.equal((await storage.updateLeasedMessage('message-1', 'recipient', winningToken, { status: 'acked' })).status, 'acked'); }); + +test('MechStorage burns a single-use key exactly once with CAS', async () => { + const storage = new MechStorage({ appId: 'test', apiKey: 'test', sdk: fakeSdk() }); + await storage.createIssuedKey({ key_id: 'key-1', key_hash: 'hash-1', used_at: null }); + const [first, second] = await Promise.all([ + storage.burnSingleUseKey('key-1'), + storage.burnSingleUseKey('key-1') + ]); + assert.deepEqual([first, second].sort(), [false, true]); + assert.ok((await storage.getIssuedKey('key-1')).used_at); +}); From 56c91b39b7a4b110497c0f128b24f2786013bf60 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:31:37 -0500 Subject: [PATCH 07/10] docs(inbox): document fenced lease tokens --- docs/ARCHITECTURE.md | 6 +++--- docs/ERROR-CODES.md | 2 ++ llms.txt | 7 ++++--- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 6335806..b2fc343 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -229,8 +229,8 @@ sequenceDiagram **NACK with lease extension:** ``` -Recipient -> Server: POST /.../nack {extend_sec: 120} -Server -> Storage: updateMessage(id, {lease_until: base + 120s}) +Recipient -> Server: POST /.../nack {extend_sec: 120, lease_token} +Server -> Storage: conditional update (id, agent, lease_token, {lease_until: base + 120s}) ``` **Reply (correlated response):** @@ -344,7 +344,7 @@ src/ storage/ index.js # Backend selector (STORAGE_BACKEND env var) memory.js # In-process Maps for development/testing - mech.js # HTTP client for Mech Storage API (production) + mech.js # Mech Storage SDK CAS adapter (production) utils/ crypto.js # Ed25519 keypair generation (tweetnacl), HKDF-SHA256, diff --git a/docs/ERROR-CODES.md b/docs/ERROR-CODES.md index aaaf050..0041d2b 100644 --- a/docs/ERROR-CODES.md +++ b/docs/ERROR-CODES.md @@ -84,6 +84,8 @@ Complete reference of all error codes returned by the Agent Dispatch Messaging P | `PULL_FAILED` | 400 | Yes | Inbox pull failed | Verify agent exists and has messages | | `ACK_FAILED` | 400 | No | Message acknowledgment failed | Ensure message is leased to this agent and in `leased` status | | `NACK_FAILED` | 400 | No | Message negative ack failed | Ensure message is leased to this agent | +| `LEASE_TOKEN_REQUIRED` | 409 | No | Ack or nack omitted the per-pull lease token | Send the exact opaque `lease_token` returned by the latest pull | +| `LEASE_FENCED` | 409 | No | Lease token is stale, belongs to another claim, or the lease expired | Pull the message again and process only the newly claimed lease | | `REPLY_FAILED` | 400 | No | Reply failed | Verify original message exists | | `MESSAGE_NOT_FOUND` | 404 | No | Message ID not found | Message may have been acked or expired | | `MESSAGE_EXPIRED` | 410 | No | Message purged (ephemeral or TTL) | Message data is gone permanently | diff --git a/llms.txt b/llms.txt index 46bb33b..11ed150 100644 --- a/llms.txt +++ b/llms.txt @@ -87,17 +87,18 @@ GET /api/agents/:agentId/identity Verification status POST /api/agents/:agentId/messages Send message [API Key — any registered agent] Body: {...envelope, ephemeral?, ttl?} -> {message_id, status} POST /api/agents/:agentId/inbox/pull Pull with lease [HTTP Sig] - Body: {visibility_timeout?} -> {message_id, envelope, lease_until, attempts} | 204 + Body: {visibility_timeout?} -> {message_id, envelope, lease_until, lease_token, attempts} | 204 POST /api/agents/:agentId/messages/:msgId/ack Acknowledge [HTTP Sig] - Body: {result?} + Body: {result?, lease_token} (required) POST /api/agents/:agentId/messages/:msgId/nack Negative ack [HTTP Sig] - Body: {extend_sec?, requeue?} + Body: {extend_sec?, requeue?, lease_token} (required) POST /api/agents/:agentId/messages/:msgId/reply Reply [HTTP Sig] Body: {...envelope} GET /api/messages/:msgId/status Delivery status [API Key] GET /api/agents/:agentId/inbox/stats Queue counts [HTTP Sig] POST /api/agents/:agentId/inbox/reclaim Reclaim expired leases [HTTP Sig] ``` +`lease_token` is an opaque, per-pull capability. Ack/nack without it return `409 LEASE_TOKEN_REQUIRED`; a stale or reclaimed token returns `409 LEASE_FENCED`. ### Groups [Agent Auth] ``` From f995233d6b9f21f75a497f876eb4f570bf413d38 Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:37:39 -0500 Subject: [PATCH 08/10] fix(storage): preserve create and membership semantics --- src/storage/mech.js | 62 +++++++++++++++++++++------------------- src/storage/mech.test.js | 15 ++++++++++ 2 files changed, 48 insertions(+), 29 deletions(-) diff --git a/src/storage/mech.js b/src/storage/mech.js index 4357b52..d6d3c2b 100644 --- a/src/storage/mech.js +++ b/src/storage/mech.js @@ -33,12 +33,23 @@ export class MechStorage { document_key: documentKey, data }); - if (!result.ok) { - const error = new Error(`Mech document already exists: ${collection}/${documentKey}`); - error.code = result.code; - throw error; + if (result.ok) return result.document.data; + + // Physical delete is not available in the SDK contract. Reuse a logical + // tombstone only when its current revision can be conditionally revived. + const existing = await this.nosql.cas.getDocument(collection, documentKey); + if (existing.ok && existing.document.data?._deleted) { + const revived = await this.nosql.cas.updateDocument(collection, documentKey, { + _rev: existing.document._rev, + data, + metadata: existing.document.metadata + }); + if (revived.ok) return revived.document.data; } - return result.document.data; + + const error = new Error(`Mech document already exists: ${collection}/${documentKey}`); + error.code = result.code; + throw error; } async getDocument(collection, documentKey) { @@ -86,6 +97,7 @@ export class MechStorage { // All updates are complete-record CAS writes. A conflict is retried from the // returned current revision so unrelated concurrent fields are preserved. let current = await this.getDocument(collection, documentKey); + if (!current) return null; for (let attempt = 0; current && attempt < 3; attempt++) { const data = { ...current.data, ...updates }; const result = await this.nosql.cas.updateDocument(collection, documentKey, { @@ -681,37 +693,29 @@ export class MechStorage { // ============ GROUP MEMBERS ============ async addGroupMember(groupId, member) { - const group = await this.getGroup(groupId); - if (!group) { - throw new Error(`Group ${groupId} not found`); - } - - const members = group.members || []; - - // Check if already a member - if (members.some(m => m.agent_id === member.agent_id)) { - throw new Error(`Agent ${member.agent_id} is already a member`); - } - const newMember = { ...member, joined_at: Date.now() }; - - members.push(newMember); - - return this.updateGroup(groupId, { members }); + const updated = await this.transitionDocument('admp_groups', groupId, group => { + const members = group.members || []; + if (members.some(existing => existing.agent_id === member.agent_id)) { + throw new Error(`Agent ${member.agent_id} is already a member`); + } + return { ...group, members: [...members, newMember], updated_at: Date.now() }; + }); + if (!updated) throw new Error(`Group ${groupId} not found`); + return updated; } async removeGroupMember(groupId, agentId) { - const group = await this.getGroup(groupId); - if (!group) { - throw new Error(`Group ${groupId} not found`); - } - - const members = (group.members || []).filter(m => m.agent_id !== agentId); - - return this.updateGroup(groupId, { members }); + const updated = await this.transitionDocument('admp_groups', groupId, group => ({ + ...group, + members: (group.members || []).filter(member => member.agent_id !== agentId), + updated_at: Date.now() + })); + if (!updated) throw new Error(`Group ${groupId} not found`); + return updated; } async getGroupMembers(groupId) { diff --git a/src/storage/mech.test.js b/src/storage/mech.test.js index 225e496..f3776c5 100644 --- a/src/storage/mech.test.js +++ b/src/storage/mech.test.js @@ -61,6 +61,8 @@ test('MechStorage merges full records through CAS and tombstones instead of dele assert.equal(await storage.deleteAgent('agent-1'), true); assert.equal(await storage.getAgent('agent-1'), null); assert.deepEqual(await storage.listAgents(), []); + await storage.createAgent({ agent_id: 'agent-1', did: 'did:key:two' }); + assert.equal((await storage.getAgent('agent-1')).did, 'did:key:two'); }); test('MechStorage pages collections and fences concurrent message claims', async () => { @@ -102,3 +104,16 @@ test('MechStorage burns a single-use key exactly once with CAS', async () => { assert.deepEqual([first, second].sort(), [false, true]); assert.ok((await storage.getIssuedKey('key-1')).used_at); }); + +test('MechStorage conditionally creates domain config and retains concurrent group members', async () => { + const storage = new MechStorage({ appId: 'test', apiKey: 'test', sdk: fakeSdk() }); + await storage.setDomainConfig('agent-1', { domain: 'example.test' }); + assert.equal((await storage.getDomainConfig('agent-1')).domain, 'example.test'); + + await storage.createGroup({ id: 'group-1', name: 'Test', members: [] }); + await Promise.all([ + storage.addGroupMember('group-1', { agent_id: 'agent-a', role: 'member' }), + storage.addGroupMember('group-1', { agent_id: 'agent-b', role: 'member' }) + ]); + assert.deepEqual((await storage.getGroup('group-1')).members.map(member => member.agent_id).sort(), ['agent-a', 'agent-b']); +}); From 11abab8956c4fcdaf881d7b5ef5bcb3b38c86baa Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:40:18 -0500 Subject: [PATCH 09/10] fix(storage): retry contended member updates --- src/storage/mech.js | 12 ++++++++++-- src/storage/mech.test.js | 8 +++----- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/src/storage/mech.js b/src/storage/mech.js index d6d3c2b..9b49a99 100644 --- a/src/storage/mech.js +++ b/src/storage/mech.js @@ -136,6 +136,9 @@ export class MechStorage { current = result.current && !result.current.data._deleted ? result.current : await this.getDocument(collection, documentKey); + if (current && attempt + 1 < maxAttempts) { + await new Promise(resolve => setTimeout(resolve, Math.min(2 ** attempt, 25))); + } } return null; } @@ -703,8 +706,13 @@ export class MechStorage { throw new Error(`Agent ${member.agent_id} is already a member`); } return { ...group, members: [...members, newMember], updated_at: Date.now() }; - }); - if (!updated) throw new Error(`Group ${groupId} not found`); + }, 10); + if (!updated) { + if (await this.getGroup(groupId)) { + throw new Error(`Concurrent update conflicted repeatedly for group ${groupId}`); + } + throw new Error(`Group ${groupId} not found`); + } return updated; } diff --git a/src/storage/mech.test.js b/src/storage/mech.test.js index f3776c5..b629c50 100644 --- a/src/storage/mech.test.js +++ b/src/storage/mech.test.js @@ -111,9 +111,7 @@ test('MechStorage conditionally creates domain config and retains concurrent gro assert.equal((await storage.getDomainConfig('agent-1')).domain, 'example.test'); await storage.createGroup({ id: 'group-1', name: 'Test', members: [] }); - await Promise.all([ - storage.addGroupMember('group-1', { agent_id: 'agent-a', role: 'member' }), - storage.addGroupMember('group-1', { agent_id: 'agent-b', role: 'member' }) - ]); - assert.deepEqual((await storage.getGroup('group-1')).members.map(member => member.agent_id).sort(), ['agent-a', 'agent-b']); + const memberIds = ['agent-a', 'agent-b', 'agent-c', 'agent-d', 'agent-e', 'agent-f']; + await Promise.all(memberIds.map(agent_id => storage.addGroupMember('group-1', { agent_id, role: 'member' }))); + assert.deepEqual((await storage.getGroup('group-1')).members.map(member => member.agent_id).sort(), memberIds); }); From c373cbbf1f6602eba2b39e4688af6a6b023f86ac Mon Sep 17 00:00:00 2001 From: dundas Date: Sun, 16 Aug 2026 17:44:34 -0500 Subject: [PATCH 10/10] fix(ci): resolve Mech SDK from scoped registry --- .npmrc | 1 + package-lock.json | 6 ++++++ 2 files changed, 7 insertions(+) create mode 100644 .npmrc diff --git a/.npmrc b/.npmrc new file mode 100644 index 0000000..2f68648 --- /dev/null +++ b/.npmrc @@ -0,0 +1 @@ +@mech:registry=https://registry.mechdna.net/api/packages/mech/npm/ diff --git a/package-lock.json b/package-lock.json index 00bd2e5..1a20af4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,6 +9,7 @@ "version": "1.0.0", "license": "MIT", "dependencies": { + "@mech/storage-sdk": "0.3.2", "cors": "^2.8.5", "dotenv": "^17.2.3", "express": "^4.18.2", @@ -40,6 +41,11 @@ "node": ">=18" } }, + "node_modules/@mech/storage-sdk": { + "version": "0.3.2", + "resolved": "https://registry.mechdna.net/api/packages/mech/npm/%40mech%2Fstorage-sdk/-/0.3.2/storage-sdk-0.3.2.tgz", + "integrity": "sha512-hb7zXI8ckc2G9/xfjvnCRy2EBYJOiqLFnvlqRx6F9lfJOTsLfMeq3qD9+ePugfhUkXX9ODOyyHlbDH+f3DdGgw==" + }, "node_modules/@noble/hashes": { "version": "1.8.0", "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-1.8.0.tgz",