From 9d6a5b536e15cddfeb4dc8b9668d433829ef6ff6 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 3 Nov 2026 00:35:39 -0500 Subject: [PATCH 1/5] tests: the wire contract's workstream examples as engine testdata The examples, field tables, enums and error codes the Workstreams wire contract carries, one file per method and event, with the extractor that regenerates them when the contract changes. Co-Authored-By: Claude Opus 5.5 --- .../testdata/workstreams-wire/README.md | 10 + .../workstreams-wire/collections.add.json | 42 + .../testdata/workstreams-wire/contract.json | 1180 +++++++++++++++++ .../testdata/workstreams-wire/extract.py | 103 ++ .../workstreams-wire/goahead.grant.json | 39 + .../workstreams-wire/goahead.list.json | 36 + .../workstreams-wire/goahead.revoke.json | 35 + .../workstreams-wire/page.create.json | 47 + .../workstreams-wire/page.delete.json | 25 + .../testdata/workstreams-wire/page.get.json | 39 + .../workstreams-wire/page.history.json | 45 + .../testdata/workstreams-wire/page.list.json | 35 + .../testdata/workstreams-wire/page.move.json | 55 + .../workstreams-wire/page.restore.json | 41 + .../testdata/workstreams-wire/page.save.json | 56 + .../workstreams-wire/page.update.json | 37 + .../workstreams-wire/record.drop.json | 39 + .../workstreams-wire/record.keep.json | 39 + .../workstreams-wire/record.list.json | 41 + .../workstreams-wire/record.propose.json | 46 + .../workstreams-wire/record.supersede.json | 65 + .../workstreams-wire/workstream.changed.json | 22 + .../workstreams-wire/workstream.create.json | 51 + .../workstreams-wire/workstream.get.json | 69 + .../workstreams-wire/workstream.list.json | 48 + .../workstream.mark_done.json | 47 + .../workstreams-wire/workstream.plan_get.json | 28 + .../workstreams-wire/workstream.plan_set.json | 86 ++ .../workstreams-wire/workstream.reopen.json | 43 + .../workstreams-wire/workstream.update.json | 52 + 30 files changed, 2501 insertions(+) create mode 100644 cmd/codeaf/testdata/workstreams-wire/README.md create mode 100644 cmd/codeaf/testdata/workstreams-wire/collections.add.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/contract.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/extract.py create mode 100644 cmd/codeaf/testdata/workstreams-wire/goahead.grant.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/goahead.list.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/goahead.revoke.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.create.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.delete.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.get.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.history.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.list.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.move.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.restore.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.save.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/page.update.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/record.drop.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/record.keep.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/record.list.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/record.propose.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/record.supersede.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.changed.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.create.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.get.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.list.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.mark_done.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.plan_get.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.plan_set.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.reopen.json create mode 100644 cmd/codeaf/testdata/workstreams-wire/workstream.update.json diff --git a/cmd/codeaf/testdata/workstreams-wire/README.md b/cmd/codeaf/testdata/workstreams-wire/README.md new file mode 100644 index 0000000000..29bc64a7b9 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/README.md @@ -0,0 +1,10 @@ +# workstreams-wire + +The examples the desk's contract carries; regenerate when the contract changes: + + python3 -I extract.py . + +`contract.json` holds the method index, the event, the error codes, the enums and every +field table (type, required). Each other file holds one method's or event's examples. +`appworkstreams_conformance_test.go` sends the examples through the wire handler and +compares each answer's shape with them. Values may differ; shapes may not. diff --git a/cmd/codeaf/testdata/workstreams-wire/collections.add.json b/cmd/codeaf/testdata/workstreams-wire/collections.add.json new file mode 100644 index 0000000000..767a4e61ed --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/collections.add.json @@ -0,0 +1,42 @@ +{ + "method": "collections.add", + "examples": { + "params": [ + { + "id": "ws_8f2c1a", + "refs": [ + { + "kind": "factory_job", + "id": "1107", + "factory": "f_local" + } + ] + } + ], + "result": [ + { + "added": [ + { + "kind": "factory_job", + "id": "1107", + "factory": "f_local", + "added_at": "2026-11-02T14:05:00Z" + } + ], + "already": [] + } + ], + "error": [ + { + "code": "in_other_workstream", + "message": "that job belongs to another workstream", + "detail": { + "factory": "f_local", + "job": 1107, + "workstream_id": "ws_92b7", + "workstream_name": "Account migration" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/contract.json b/cmd/codeaf/testdata/workstreams-wire/contract.json new file mode 100644 index 0000000000..df9349aa4f --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/contract.json @@ -0,0 +1,1180 @@ +{ + "methods": [ + "workstream.list", + "workstream.get", + "workstream.create", + "workstream.update", + "workstream.mark_done", + "workstream.reopen", + "workstream.plan_get", + "workstream.plan_set", + "page.list", + "page.get", + "page.save", + "page.create", + "page.update", + "page.move", + "page.delete", + "page.history", + "page.restore", + "record.list", + "record.propose", + "record.keep", + "record.drop", + "record.supersede", + "goahead.list", + "goahead.grant", + "goahead.revoke" + ], + "events": [ + { + "event": "workstream.changed", + "data_type": "WorkstreamChanged" + } + ], + "errors": [ + { + "code": "bad_params", + "desk_local": false, + "detail_type": null + }, + { + "code": "not_found", + "desk_local": false, + "detail_type": null + }, + { + "code": "conflict", + "desk_local": false, + "detail_type": null + }, + { + "code": "in_other_workstream", + "desk_local": false, + "detail_type": "WorkstreamTaken" + }, + { + "code": "version_conflict", + "desk_local": false, + "detail_type": "WorkstreamVersionConflict" + }, + { + "code": "cycle", + "desk_local": false, + "detail_type": null + }, + { + "code": "too_large", + "desk_local": false, + "detail_type": null + }, + { + "code": "busy", + "desk_local": false, + "detail_type": null + }, + { + "code": "unknown_method", + "desk_local": false, + "detail_type": null + }, + { + "code": "unsupported", + "desk_local": true, + "detail_type": null + } + ], + "enums": { + "WorkstreamStage": [ + "shaping", + "building", + "done" + ], + "WorkstreamAutonomy": [ + "propose", + "small", + "off" + ], + "WorkstreamActor": [ + "you", + "agent" + ], + "WorkstreamRefKind": [ + "collection", + "conversation", + "task", + "artifact", + "crew", + "process", + "factory_job" + ], + "WorkstreamSourceKind": [ + "chat", + "message", + "factory_job", + "page", + "url" + ], + "WorkstreamPlanItemKind": [ + "job", + "draft" + ], + "WorkstreamRecordKind": [ + "decision", + "question" + ], + "WorkstreamRecordState": [ + "proposed", + "kept", + "dropped", + "superseded" + ], + "WorkstreamGoAheadScope": [ + "job", + "wave", + "milestone" + ], + "WorkstreamGoAheadState": [ + "active", + "ended" + ], + "WorkstreamGoAheadEndReason": [ + "expired", + "revoked", + "workstream_done" + ], + "WorkstreamChangedPart": [ + "workstream", + "plan", + "pages", + "records", + "goaheads", + "jobs" + ] + }, + "types": { + "WorkstreamWake": { + "morning": { + "type": "boolean", + "required": true + }, + "related_issues": { + "type": "boolean", + "required": true + } + }, + "WorkstreamRef": { + "kind": { + "type": "WorkstreamRefKind", + "required": true + }, + "id": { + "type": "string", + "required": true + }, + "session_id": { + "type": "string", + "required": false + }, + "factory": { + "type": "string", + "required": false + } + }, + "WorkstreamSource": { + "kind": { + "type": "WorkstreamSourceKind", + "required": true + }, + "id": { + "type": "string", + "required": true + }, + "chat_id": { + "type": "string", + "required": false + }, + "factory": { + "type": "string", + "required": false + }, + "label": { + "type": "string", + "required": false + } + }, + "WorkstreamRow": { + "id": { + "type": "string", + "required": true + }, + "name": { + "type": "string", + "required": true + }, + "stage": { + "type": "WorkstreamStage", + "required": true + }, + "total": { + "type": "integer", + "required": true + }, + "merged": { + "type": "integer", + "required": false + }, + "needs_you": { + "type": "integer", + "required": false + }, + "spend_usd": { + "type": "number (USD)", + "required": false + }, + "budget_usd": { + "type": "number (USD)", + "required": false + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": true + } + }, + "Workstream": { + "id": { + "type": "string", + "required": true + }, + "name": { + "type": "string", + "required": true + }, + "why": { + "type": "string", + "required": true + }, + "stage": { + "type": "WorkstreamStage", + "required": true + }, + "agent_id": { + "type": "string", + "required": true + }, + "origin_chat_id": { + "type": "string", + "required": false + }, + "autonomy": { + "type": "WorkstreamAutonomy", + "required": true + }, + "wake": { + "type": "WorkstreamWake", + "required": true + }, + "budget_usd": { + "type": "number (USD)", + "required": false + }, + "created_at": { + "type": "string (RFC 3339)", + "required": true + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": true + }, + "started_at": { + "type": "string (RFC 3339)", + "required": false + }, + "done_at": { + "type": "string (RFC 3339)", + "required": false + }, + "how_it_went": { + "type": "string", + "required": false + }, + "total": { + "type": "integer", + "required": true + }, + "merged": { + "type": "integer", + "required": false + }, + "needs_you": { + "type": "integer", + "required": false + }, + "spend_usd": { + "type": "number (USD)", + "required": false + } + }, + "WorkstreamPlan": { + "workstream_id": { + "type": "string", + "required": true + }, + "version": { + "type": "integer", + "required": true + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": false + }, + "milestones": { + "type": "WorkstreamMilestone[]", + "required": true + }, + "waves": { + "type": "WorkstreamWave[]", + "required": true + } + }, + "WorkstreamMilestone": { + "id": { + "type": "string", + "required": true + }, + "title": { + "type": "string", + "required": true + } + }, + "WorkstreamWave": { + "n": { + "type": "integer", + "required": true + }, + "title": { + "type": "string", + "required": false + }, + "milestone_id": { + "type": "string", + "required": false + }, + "auto_start": { + "type": "boolean", + "required": true + }, + "items": { + "type": "WorkstreamPlanItem[]", + "required": true + } + }, + "WorkstreamPlanItem": { + "kind": { + "type": "WorkstreamPlanItemKind", + "required": true + }, + "factory": { + "type": "string", + "required": false + }, + "job": { + "type": "integer", + "required": false + }, + "draft_id": { + "type": "string", + "required": false + }, + "title": { + "type": "string", + "required": false + }, + "repo": { + "type": "string", + "required": false + }, + "text": { + "type": "string", + "required": false + } + }, + "WorkstreamPageNode": { + "id": { + "type": "string", + "required": true + }, + "workstream_id": { + "type": "string", + "required": true + }, + "parent_id": { + "type": "string", + "required": false + }, + "title": { + "type": "string", + "required": true + }, + "owner": { + "type": "WorkstreamActor", + "required": true + }, + "read_by_jobs": { + "type": "boolean", + "required": true + }, + "position": { + "type": "integer", + "required": true + }, + "version": { + "type": "integer", + "required": true + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": true + }, + "updated_by": { + "type": "WorkstreamActor", + "required": true + }, + "updated_from": { + "type": "WorkstreamSource", + "required": false + } + }, + "WorkstreamPage": { + "id": { + "type": "string", + "required": true + }, + "workstream_id": { + "type": "string", + "required": true + }, + "parent_id": { + "type": "string", + "required": false + }, + "title": { + "type": "string", + "required": true + }, + "owner": { + "type": "WorkstreamActor", + "required": true + }, + "read_by_jobs": { + "type": "boolean", + "required": true + }, + "position": { + "type": "integer", + "required": true + }, + "version": { + "type": "integer", + "required": true + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": true + }, + "updated_by": { + "type": "WorkstreamActor", + "required": true + }, + "updated_from": { + "type": "WorkstreamSource", + "required": false + }, + "doc": { + "type": "object", + "required": true + }, + "markdown": { + "type": "string", + "required": false + } + }, + "WorkstreamPageVersion": { + "version": { + "type": "integer", + "required": true + }, + "saved_at": { + "type": "string (RFC 3339)", + "required": true + }, + "saved_by": { + "type": "WorkstreamActor", + "required": true + }, + "from": { + "type": "WorkstreamSource", + "required": false + }, + "title": { + "type": "string", + "required": true + }, + "bytes": { + "type": "integer", + "required": true + } + }, + "WorkstreamRecord": { + "id": { + "type": "string", + "required": true + }, + "workstream_id": { + "type": "string", + "required": true + }, + "kind": { + "type": "WorkstreamRecordKind", + "required": true + }, + "text": { + "type": "string", + "required": true + }, + "state": { + "type": "WorkstreamRecordState", + "required": true + }, + "by": { + "type": "WorkstreamActor", + "required": true + }, + "source": { + "type": "WorkstreamSource", + "required": false + }, + "created_at": { + "type": "string (RFC 3339)", + "required": true + }, + "decided_at": { + "type": "string (RFC 3339)", + "required": false + }, + "supersedes": { + "type": "string", + "required": false + }, + "superseded_by": { + "type": "string", + "required": false + } + }, + "WorkstreamGoAhead": { + "id": { + "type": "string", + "required": true + }, + "workstream_id": { + "type": "string", + "required": true + }, + "scope": { + "type": "WorkstreamGoAheadScope", + "required": true + }, + "factory": { + "type": "string", + "required": false + }, + "job": { + "type": "integer", + "required": false + }, + "wave": { + "type": "integer", + "required": false + }, + "milestone_id": { + "type": "string", + "required": false + }, + "cap_usd": { + "type": "number (USD)", + "required": true + }, + "ends_at": { + "type": "string (RFC 3339)", + "required": true + }, + "state": { + "type": "WorkstreamGoAheadState", + "required": true + }, + "end_reason": { + "type": "WorkstreamGoAheadEndReason", + "required": false + }, + "spent_usd": { + "type": "number (USD)", + "required": false + }, + "granted_at": { + "type": "string (RFC 3339)", + "required": true + }, + "ended_at": { + "type": "string (RFC 3339)", + "required": false + } + }, + "WorkstreamChanged": { + "id": { + "type": "string", + "required": true + }, + "parts": { + "type": "WorkstreamChangedPart[]", + "required": false + } + }, + "WorkstreamTaken": { + "factory": { + "type": "string", + "required": true + }, + "job": { + "type": "integer", + "required": true + }, + "workstream_id": { + "type": "string", + "required": true + }, + "workstream_name": { + "type": "string", + "required": true + } + }, + "WorkstreamVersionConflict": { + "id": { + "type": "string", + "required": true + }, + "version": { + "type": "integer", + "required": true + }, + "updated_at": { + "type": "string (RFC 3339)", + "required": true + }, + "updated_by": { + "type": "WorkstreamActor", + "required": true + } + }, + "WorkstreamPatch": { + "name": { + "type": "string", + "required": false + }, + "why": { + "type": "string", + "required": false + }, + "autonomy": { + "type": "WorkstreamAutonomy", + "required": false + }, + "wake": { + "type": "WorkstreamWake", + "required": false + }, + "budget_usd": { + "type": "number (USD)", + "required": false + } + }, + "WorkstreamListParams": {}, + "WorkstreamListResult": { + "workstreams": { + "type": "WorkstreamRow[]", + "required": true + } + }, + "WorkstreamGetParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamGetResult": { + "workstream": { + "type": "Workstream", + "required": true + } + }, + "WorkstreamCreateParams": { + "name": { + "type": "string", + "required": true + }, + "why": { + "type": "string", + "required": false + }, + "origin_chat_id": { + "type": "string", + "required": false + }, + "refs": { + "type": "WorkstreamRef[]", + "required": false + }, + "model": { + "type": "string", + "required": false + }, + "client_id": { + "type": "string", + "required": false + } + }, + "WorkstreamCreateResult": { + "workstream": { + "type": "Workstream", + "required": true + } + }, + "WorkstreamUpdateParams": { + "id": { + "type": "string", + "required": true + }, + "patch": { + "type": "WorkstreamPatch", + "required": true + } + }, + "WorkstreamUpdateResult": { + "workstream": { + "type": "Workstream", + "required": true + } + }, + "WorkstreamMarkDoneParams": { + "id": { + "type": "string", + "required": true + }, + "how_it_went": { + "type": "string", + "required": false + }, + "archive_chats": { + "type": "boolean", + "required": false + } + }, + "WorkstreamMarkDoneResult": { + "workstream": { + "type": "Workstream", + "required": true + } + }, + "WorkstreamReopenParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamReopenResult": { + "workstream": { + "type": "Workstream", + "required": true + } + }, + "WorkstreamPlanGetParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamPlanGetResult": { + "plan": { + "type": "WorkstreamPlan", + "required": true + } + }, + "WorkstreamPlanSetParams": { + "id": { + "type": "string", + "required": true + }, + "base_version": { + "type": "integer", + "required": true + }, + "milestones": { + "type": "WorkstreamMilestone[]", + "required": true + }, + "waves": { + "type": "WorkstreamWave[]", + "required": true + } + }, + "WorkstreamPlanSetResult": { + "plan": { + "type": "WorkstreamPlan", + "required": true + } + }, + "WorkstreamPageListParams": { + "workstream_id": { + "type": "string", + "required": true + } + }, + "WorkstreamPageListResult": { + "pages": { + "type": "WorkstreamPageNode[]", + "required": true + } + }, + "WorkstreamPageGetParams": { + "id": { + "type": "string", + "required": true + }, + "markdown": { + "type": "boolean", + "required": false + } + }, + "WorkstreamPageGetResult": { + "page": { + "type": "WorkstreamPage", + "required": true + } + }, + "WorkstreamPageSaveParams": { + "id": { + "type": "string", + "required": true + }, + "doc": { + "type": "object", + "required": true + }, + "base_version": { + "type": "integer", + "required": true + } + }, + "WorkstreamPageSaveResult": { + "page": { + "type": "WorkstreamPage", + "required": true + } + }, + "WorkstreamPageCreateParams": { + "workstream_id": { + "type": "string", + "required": true + }, + "parent_id": { + "type": "string", + "required": false + }, + "after_id": { + "type": "string", + "required": false + }, + "title": { + "type": "string", + "required": true + }, + "owner": { + "type": "WorkstreamActor", + "required": false + }, + "read_by_jobs": { + "type": "boolean", + "required": false + }, + "doc": { + "type": "object", + "required": false + } + }, + "WorkstreamPageCreateResult": { + "page": { + "type": "WorkstreamPage", + "required": true + } + }, + "WorkstreamPageUpdateParams": { + "id": { + "type": "string", + "required": true + }, + "title": { + "type": "string", + "required": false + }, + "owner": { + "type": "WorkstreamActor", + "required": false + }, + "read_by_jobs": { + "type": "boolean", + "required": false + } + }, + "WorkstreamPageUpdateResult": { + "page": { + "type": "WorkstreamPageNode", + "required": true + } + }, + "WorkstreamPageMoveParams": { + "id": { + "type": "string", + "required": true + }, + "parent_id": { + "type": "string", + "required": false + }, + "after_id": { + "type": "string", + "required": false + } + }, + "WorkstreamPageMoveResult": { + "page": { + "type": "WorkstreamPageNode", + "required": true + } + }, + "WorkstreamPageDeleteParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamPageDeleteResult": { + "deleted": { + "type": "string[]", + "required": true + } + }, + "WorkstreamPageHistoryParams": { + "id": { + "type": "string", + "required": true + }, + "limit": { + "type": "integer", + "required": false + } + }, + "WorkstreamPageHistoryResult": { + "versions": { + "type": "WorkstreamPageVersion[]", + "required": true + } + }, + "WorkstreamPageRestoreParams": { + "id": { + "type": "string", + "required": true + }, + "version": { + "type": "integer", + "required": true + }, + "base_version": { + "type": "integer", + "required": true + } + }, + "WorkstreamPageRestoreResult": { + "page": { + "type": "WorkstreamPage", + "required": true + } + }, + "WorkstreamRecordListParams": { + "workstream_id": { + "type": "string", + "required": true + }, + "kind": { + "type": "WorkstreamRecordKind", + "required": false + }, + "state": { + "type": "WorkstreamRecordState", + "required": false + } + }, + "WorkstreamRecordListResult": { + "records": { + "type": "WorkstreamRecord[]", + "required": true + } + }, + "WorkstreamRecordProposeParams": { + "workstream_id": { + "type": "string", + "required": true + }, + "kind": { + "type": "WorkstreamRecordKind", + "required": true + }, + "text": { + "type": "string", + "required": true + }, + "source": { + "type": "WorkstreamSource", + "required": false + }, + "kept": { + "type": "boolean", + "required": false + } + }, + "WorkstreamRecordProposeResult": { + "record": { + "type": "WorkstreamRecord", + "required": true + } + }, + "WorkstreamRecordKeepParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamRecordKeepResult": { + "record": { + "type": "WorkstreamRecord", + "required": true + } + }, + "WorkstreamRecordDropParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamRecordDropResult": { + "record": { + "type": "WorkstreamRecord", + "required": true + } + }, + "WorkstreamRecordSupersedeParams": { + "id": { + "type": "string", + "required": true + }, + "text": { + "type": "string", + "required": true + }, + "kind": { + "type": "WorkstreamRecordKind", + "required": false + }, + "source": { + "type": "WorkstreamSource", + "required": false + } + }, + "WorkstreamRecordSupersedeResult": { + "record": { + "type": "WorkstreamRecord", + "required": true + }, + "superseded": { + "type": "WorkstreamRecord", + "required": true + } + }, + "WorkstreamGoAheadListParams": { + "workstream_id": { + "type": "string", + "required": true + }, + "ended": { + "type": "boolean", + "required": false + } + }, + "WorkstreamGoAheadListResult": { + "goaheads": { + "type": "WorkstreamGoAhead[]", + "required": true + } + }, + "WorkstreamGoAheadGrantParams": { + "workstream_id": { + "type": "string", + "required": true + }, + "scope": { + "type": "WorkstreamGoAheadScope", + "required": true + }, + "factory": { + "type": "string", + "required": false + }, + "job": { + "type": "integer", + "required": false + }, + "wave": { + "type": "integer", + "required": false + }, + "milestone_id": { + "type": "string", + "required": false + }, + "cap_usd": { + "type": "number (USD)", + "required": true + }, + "ends_at": { + "type": "string (RFC 3339)", + "required": true + } + }, + "WorkstreamGoAheadGrantResult": { + "goahead": { + "type": "WorkstreamGoAhead", + "required": true + } + }, + "WorkstreamGoAheadRevokeParams": { + "id": { + "type": "string", + "required": true + } + }, + "WorkstreamGoAheadRevokeResult": { + "goahead": { + "type": "WorkstreamGoAhead", + "required": true + } + } + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/extract.py b/cmd/codeaf/testdata/workstreams-wire/extract.py new file mode 100644 index 0000000000..546b10b286 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/extract.py @@ -0,0 +1,103 @@ +"""Extract wire shapes (types, enums, methods, examples) from the workstreams wire contract. + +usage: python3 -I extract.py +Writes contract.json (index, enums, errors, types) and one .json per method/event +with its examples. No prose is copied. +""" +import json, os, re, sys + +src, out = sys.argv[1], sys.argv[2] +lines = open(src, encoding="utf-8").read().split("\n") +end = next(i for i, l in enumerate(lines) if l.startswith("## Page document format")) +lines = lines[:end] + +TICK = re.compile(r"`([^`]*)`") + +def table_at(i): + """rows of the markdown table starting at or after line i (skip blank lines).""" + while i < len(lines) and not lines[i].startswith("|"): + i += 1 + rows = [] + while i < len(lines) and lines[i].startswith("|"): + cells = [c.strip() for c in lines[i].strip().strip("|").split("|")] + rows.append(cells) + i += 1 + return rows[2:] # drop header + separator + +def fields(rows): + out = {} + for cells in rows: + if cells[0] in ("—", "-"): + continue + name = TICK.fullmatch(cells[0]).group(1) + typ = TICK.fullmatch(cells[1]).group(1) + req = cells[2] + assert req in ("yes", "no"), cells + out[name] = {"type": typ, "required": req == "yes"} + return out + +types, enums, errors, methods, events = {}, {}, [], [], [] +method_info = {} +examples = {} +section = None # H2 +h3 = None +i = 0 +while i < len(lines): + l = lines[i] + if l.startswith("## "): + section, h3 = l[3:].strip(), None + elif l.startswith("### "): + h3 = TICK.search(l).group(1) + if section == "Shared shapes": + types[h3] = fields(table_at(i + 1)) + elif section == "Methods": + methods.append(h3) + method_info[h3] = {"method": h3, "errors": []} + elif section == "Shared shapes" and l.startswith("| type | words |"): + for cells in table_at(i): + enums[TICK.fullmatch(cells[0]).group(1)] = TICK.findall(cells[1]) + elif section == "Method index" and l.startswith("| event | data |"): + for cells in table_at(i): + events.append({"event": TICK.fullmatch(cells[0]).group(1), "data_type": TICK.fullmatch(cells[1]).group(1)}) + elif section == "Method index" and l.startswith("| method | params | result |"): + index = [TICK.fullmatch(c[0]).group(1) for c in table_at(i)] + elif section == "Error codes" and l.startswith("| code |"): + for cells in table_at(i): + m = re.match(r"`(\w+)`( \(desk-local\))?", cells[0]) + det = re.match(r"`(Workstream\w+)`", cells[2]) + errors.append({"code": m.group(1), "desk_local": bool(m.group(2)), "detail_type": det.group(1) if det else None}) + m = re.match(r"^(Params|Result) \(`(\w+)`\):", l) + if m and section == "Methods": + types[m.group(2)] = fields(table_at(i + 1)) + method_info[h3][m.group(1).lower() + "_type"] = m.group(2) + if l.startswith("Errors:") and section == "Methods": + method_info[h3]["errors"] = TICK.findall(l) + m = re.match(r"^```json (\S+) (params|result|error|data)$", l) + if m: + j = i + 1 + while lines[j] != "```": + j += 1 + body = json.loads("\n".join(lines[i + 1:j])) + examples.setdefault(m.group(1), {}).setdefault(m.group(2), []).append(body) + i = j + i += 1 + +assert index == methods, (index, methods) +os.makedirs(out, exist_ok=True) +contract = {"methods": methods, "events": events, "errors": errors, "enums": enums, "types": types} +def dump(path, v): + with open(os.path.join(out, path), "w", encoding="utf-8") as f: + json.dump(v, f, indent=2, ensure_ascii=False) + f.write("\n") +dump("contract.json", contract) +for name in methods: + info = dict(method_info[name]) + info["examples"] = examples.pop(name) + dump(name + ".json", info) +for ev in events: + dump(ev["event"] + ".json", {"event": ev["event"], "data_type": ev["data_type"], "examples": examples.pop(ev["event"])}) +# collection examples the doc carries for the factory_job kind +for name, ex in sorted(examples.items()): + dump(name + ".json", {"method": name, "examples": ex}) + print("extra:", name, sorted(ex)) +print(len(methods), "methods;", len(types), "types;", len(enums), "enums;", len(errors), "error codes") diff --git a/cmd/codeaf/testdata/workstreams-wire/goahead.grant.json b/cmd/codeaf/testdata/workstreams-wire/goahead.grant.json new file mode 100644 index 0000000000..08fd83ecc5 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/goahead.grant.json @@ -0,0 +1,39 @@ +{ + "method": "goahead.grant", + "errors": [ + "bad_params", + "not_found", + "conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamGoAheadGrantParams", + "result_type": "WorkstreamGoAheadGrantResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a", + "scope": "job", + "factory": "f_local", + "job": 1107, + "cap_usd": 40.0, + "ends_at": "2026-11-03T14:05:00Z" + } + ], + "result": [ + { + "goahead": { + "id": "ga_0c44", + "workstream_id": "ws_8f2c1a", + "scope": "job", + "factory": "f_local", + "job": 1107, + "cap_usd": 40.0, + "ends_at": "2026-11-03T14:05:00Z", + "state": "active", + "granted_at": "2026-11-02T14:05:00Z" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/goahead.list.json b/cmd/codeaf/testdata/workstreams-wire/goahead.list.json new file mode 100644 index 0000000000..7a6fe79d3c --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/goahead.list.json @@ -0,0 +1,36 @@ +{ + "method": "goahead.list", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamGoAheadListParams", + "result_type": "WorkstreamGoAheadListResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a", + "ended": true + } + ], + "result": [ + { + "goaheads": [ + { + "id": "ga_0c44", + "workstream_id": "ws_8f2c1a", + "scope": "job", + "factory": "f_local", + "job": 1107, + "cap_usd": 40.0, + "ends_at": "2026-11-03T14:05:00Z", + "state": "active", + "granted_at": "2026-11-02T14:05:00Z" + } + ] + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/goahead.revoke.json b/cmd/codeaf/testdata/workstreams-wire/goahead.revoke.json new file mode 100644 index 0000000000..db4c6ac2b4 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/goahead.revoke.json @@ -0,0 +1,35 @@ +{ + "method": "goahead.revoke", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamGoAheadRevokeParams", + "result_type": "WorkstreamGoAheadRevokeResult", + "examples": { + "params": [ + { + "id": "ga_0c44" + } + ], + "result": [ + { + "goahead": { + "id": "ga_0c44", + "workstream_id": "ws_8f2c1a", + "scope": "job", + "factory": "f_local", + "job": 1107, + "cap_usd": 40.0, + "ends_at": "2026-11-03T14:05:00Z", + "state": "ended", + "granted_at": "2026-11-02T14:05:00Z", + "end_reason": "revoked", + "ended_at": "2026-11-02T15:00:00Z" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.create.json b/cmd/codeaf/testdata/workstreams-wire/page.create.json new file mode 100644 index 0000000000..10deddafb8 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.create.json @@ -0,0 +1,47 @@ +{ + "method": "page.create", + "errors": [ + "bad_params", + "not_found", + "too_large", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageCreateParams", + "result_type": "WorkstreamPageCreateResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a", + "parent_id": "pg_31d0", + "title": "Metering decisions", + "owner": "you", + "read_by_jobs": false, + "doc": { + "type": "doc", + "content": [] + } + } + ], + "result": [ + { + "page": { + "id": "pg_32e1", + "workstream_id": "ws_8f2c1a", + "title": "Metering decisions", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "updated_by": "you", + "parent_id": "pg_31d0", + "doc": { + "type": "doc", + "content": [] + } + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.delete.json b/cmd/codeaf/testdata/workstreams-wire/page.delete.json new file mode 100644 index 0000000000..f13834bec0 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.delete.json @@ -0,0 +1,25 @@ +{ + "method": "page.delete", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageDeleteParams", + "result_type": "WorkstreamPageDeleteResult", + "examples": { + "params": [ + { + "id": "pg_32e1" + } + ], + "result": [ + { + "deleted": [ + "pg_32e1" + ] + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.get.json b/cmd/codeaf/testdata/workstreams-wire/page.get.json new file mode 100644 index 0000000000..fdcf420091 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.get.json @@ -0,0 +1,39 @@ +{ + "method": "page.get", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageGetParams", + "result_type": "WorkstreamPageGetResult", + "examples": { + "params": [ + { + "id": "pg_31d0", + "markdown": true + } + ], + "result": [ + { + "page": { + "id": "pg_31d0", + "workstream_id": "ws_8f2c1a", + "title": "Overview", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "updated_by": "you", + "doc": { + "type": "doc", + "content": [] + }, + "markdown": "" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.history.json b/cmd/codeaf/testdata/workstreams-wire/page.history.json new file mode 100644 index 0000000000..ea5e6a6ce4 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.history.json @@ -0,0 +1,45 @@ +{ + "method": "page.history", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageHistoryParams", + "result_type": "WorkstreamPageHistoryResult", + "examples": { + "params": [ + { + "id": "pg_31d0", + "limit": 50 + } + ], + "result": [ + { + "versions": [ + { + "version": 2, + "saved_at": "2026-11-02T15:00:00Z", + "saved_by": "agent", + "from": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "title": "Overview", + "bytes": 27 + }, + { + "version": 1, + "saved_at": "2026-11-02T14:05:00Z", + "saved_by": "you", + "title": "Overview", + "bytes": 27 + } + ] + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.list.json b/cmd/codeaf/testdata/workstreams-wire/page.list.json new file mode 100644 index 0000000000..37550b606a --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.list.json @@ -0,0 +1,35 @@ +{ + "method": "page.list", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageListParams", + "result_type": "WorkstreamPageListResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a" + } + ], + "result": [ + { + "pages": [ + { + "id": "pg_31d0", + "workstream_id": "ws_8f2c1a", + "title": "Overview", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "updated_by": "you" + } + ] + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.move.json b/cmd/codeaf/testdata/workstreams-wire/page.move.json new file mode 100644 index 0000000000..2079f5087e --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.move.json @@ -0,0 +1,55 @@ +{ + "method": "page.move", + "errors": [ + "bad_params", + "not_found", + "cycle", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageMoveParams", + "result_type": "WorkstreamPageMoveResult", + "examples": { + "params": [ + { + "id": "pg_32e1", + "parent_id": "pg_31d0" + }, + { + "id": "pg_32e1", + "parent_id": "pg_31d0", + "after_id": "pg_548c" + } + ], + "result": [ + { + "page": { + "id": "pg_32e1", + "workstream_id": "ws_8f2c1a", + "title": "Metering decisions", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "updated_by": "you", + "parent_id": "pg_31d0" + } + }, + { + "page": { + "id": "pg_32e1", + "workstream_id": "ws_8f2c1a", + "title": "Metering decisions", + "owner": "you", + "read_by_jobs": false, + "position": 1, + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "updated_by": "you", + "parent_id": "pg_31d0" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.restore.json b/cmd/codeaf/testdata/workstreams-wire/page.restore.json new file mode 100644 index 0000000000..9cdf549c70 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.restore.json @@ -0,0 +1,41 @@ +{ + "method": "page.restore", + "errors": [ + "bad_params", + "not_found", + "version_conflict", + "too_large", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageRestoreParams", + "result_type": "WorkstreamPageRestoreResult", + "examples": { + "params": [ + { + "id": "pg_31d0", + "version": 1, + "base_version": 2 + } + ], + "result": [ + { + "page": { + "id": "pg_31d0", + "workstream_id": "ws_8f2c1a", + "title": "Overview", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 3, + "updated_at": "2026-11-02T15:00:00Z", + "updated_by": "you", + "doc": { + "type": "doc", + "content": [] + } + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.save.json b/cmd/codeaf/testdata/workstreams-wire/page.save.json new file mode 100644 index 0000000000..914a68b84e --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.save.json @@ -0,0 +1,56 @@ +{ + "method": "page.save", + "errors": [ + "bad_params", + "not_found", + "version_conflict", + "too_large", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageSaveParams", + "result_type": "WorkstreamPageSaveResult", + "examples": { + "params": [ + { + "id": "pg_31d0", + "doc": { + "type": "doc", + "content": [] + }, + "base_version": 1 + } + ], + "result": [ + { + "page": { + "id": "pg_31d0", + "workstream_id": "ws_8f2c1a", + "title": "Overview", + "owner": "you", + "read_by_jobs": false, + "position": 0, + "version": 2, + "updated_at": "2026-11-02T15:00:00Z", + "updated_by": "you", + "doc": { + "type": "doc", + "content": [] + } + } + } + ], + "error": [ + { + "code": "version_conflict", + "message": "the page changed since it was read", + "detail": { + "id": "pg_31d0", + "version": 2, + "updated_at": "2026-11-02T15:00:00Z", + "updated_by": "agent" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/page.update.json b/cmd/codeaf/testdata/workstreams-wire/page.update.json new file mode 100644 index 0000000000..1a63b0e925 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/page.update.json @@ -0,0 +1,37 @@ +{ + "method": "page.update", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPageUpdateParams", + "result_type": "WorkstreamPageUpdateResult", + "examples": { + "params": [ + { + "id": "pg_32e1", + "title": "Metering decisions", + "owner": "agent", + "read_by_jobs": true + } + ], + "result": [ + { + "page": { + "id": "pg_32e1", + "workstream_id": "ws_8f2c1a", + "title": "Metering decisions", + "owner": "agent", + "read_by_jobs": true, + "position": 0, + "version": 1, + "updated_at": "2026-11-02T15:00:00Z", + "updated_by": "you", + "parent_id": "pg_31d0" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/record.drop.json b/cmd/codeaf/testdata/workstreams-wire/record.drop.json new file mode 100644 index 0000000000..4af5590af0 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/record.drop.json @@ -0,0 +1,39 @@ +{ + "method": "record.drop", + "errors": [ + "bad_params", + "not_found", + "conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamRecordDropParams", + "result_type": "WorkstreamRecordDropResult", + "examples": { + "params": [ + { + "id": "rec_7a1e" + } + ], + "result": [ + { + "record": { + "id": "rec_7a1e", + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "state": "dropped", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T14:05:00Z", + "decided_at": "2026-11-02T15:00:00Z" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/record.keep.json b/cmd/codeaf/testdata/workstreams-wire/record.keep.json new file mode 100644 index 0000000000..faffcccb6b --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/record.keep.json @@ -0,0 +1,39 @@ +{ + "method": "record.keep", + "errors": [ + "bad_params", + "not_found", + "conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamRecordKeepParams", + "result_type": "WorkstreamRecordKeepResult", + "examples": { + "params": [ + { + "id": "rec_7a1e" + } + ], + "result": [ + { + "record": { + "id": "rec_7a1e", + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "state": "kept", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T14:05:00Z", + "decided_at": "2026-11-02T15:00:00Z" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/record.list.json b/cmd/codeaf/testdata/workstreams-wire/record.list.json new file mode 100644 index 0000000000..06b669638b --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/record.list.json @@ -0,0 +1,41 @@ +{ + "method": "record.list", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamRecordListParams", + "result_type": "WorkstreamRecordListResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a", + "kind": "question", + "state": "proposed" + } + ], + "result": [ + { + "records": [ + { + "id": "rec_7a1e", + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "state": "proposed", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T14:05:00Z" + } + ] + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/record.propose.json b/cmd/codeaf/testdata/workstreams-wire/record.propose.json new file mode 100644 index 0000000000..1ef07f72ea --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/record.propose.json @@ -0,0 +1,46 @@ +{ + "method": "record.propose", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamRecordProposeParams", + "result_type": "WorkstreamRecordProposeResult", + "examples": { + "params": [ + { + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "kept": false + } + ], + "result": [ + { + "record": { + "id": "rec_7a1e", + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "state": "proposed", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T14:05:00Z" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/record.supersede.json b/cmd/codeaf/testdata/workstreams-wire/record.supersede.json new file mode 100644 index 0000000000..cbc5623e64 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/record.supersede.json @@ -0,0 +1,65 @@ +{ + "method": "record.supersede", + "errors": [ + "bad_params", + "not_found", + "conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamRecordSupersedeParams", + "result_type": "WorkstreamRecordSupersedeResult", + "examples": { + "params": [ + { + "id": "rec_7a1e", + "text": "Retries do not count toward usage.", + "kind": "decision", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + } + } + ], + "result": [ + { + "record": { + "id": "rec_91bf", + "workstream_id": "ws_8f2c1a", + "kind": "decision", + "text": "Retries do not count toward usage.", + "state": "kept", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T15:00:00Z", + "decided_at": "2026-11-02T15:00:00Z", + "supersedes": "rec_7a1e" + }, + "superseded": { + "id": "rec_7a1e", + "workstream_id": "ws_8f2c1a", + "kind": "question", + "text": "Should retries count toward usage?", + "state": "superseded", + "by": "agent", + "source": { + "kind": "message", + "id": "msg_19a2", + "chat_id": "conv_5b9e", + "label": "Billing scope" + }, + "created_at": "2026-11-02T14:05:00Z", + "decided_at": "2026-11-02T15:00:00Z", + "superseded_by": "rec_91bf" + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.changed.json b/cmd/codeaf/testdata/workstreams-wire/workstream.changed.json new file mode 100644 index 0000000000..663445fa90 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.changed.json @@ -0,0 +1,22 @@ +{ + "event": "workstream.changed", + "data_type": "WorkstreamChanged", + "examples": { + "data": [ + { + "id": "ws_8f2c1a", + "parts": [ + "workstream", + "plan", + "pages", + "records", + "goaheads", + "jobs" + ] + }, + { + "id": "ws_8f2c1a" + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.create.json b/cmd/codeaf/testdata/workstreams-wire/workstream.create.json new file mode 100644 index 0000000000..8b400ede0c --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.create.json @@ -0,0 +1,51 @@ +{ + "method": "workstream.create", + "errors": [ + "bad_params", + "not_found", + "conflict", + "in_other_workstream", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamCreateParams", + "result_type": "WorkstreamCreateResult", + "examples": { + "params": [ + { + "name": "Usage-based billing", + "why": "Charge for recorded usage.", + "origin_chat_id": "conv_5b9e", + "refs": [ + { + "kind": "factory_job", + "id": "1107", + "factory": "f_local" + } + ], + "model": "openai/gpt-5.2", + "client_id": "create_684a" + } + ], + "result": [ + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "Charge for recorded usage.", + "stage": "shaping", + "agent_id": "conv_8d22", + "origin_chat_id": "conv_5b9e", + "autonomy": "propose", + "wake": { + "morning": true, + "related_issues": false + }, + "created_at": "2026-11-02T14:05:00Z", + "updated_at": "2026-11-02T14:05:00Z", + "total": 1 + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.get.json b/cmd/codeaf/testdata/workstreams-wire/workstream.get.json new file mode 100644 index 0000000000..cc8f857e22 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.get.json @@ -0,0 +1,69 @@ +{ + "method": "workstream.get", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamGetParams", + "result_type": "WorkstreamGetResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a" + } + ], + "result": [ + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "Charge for recorded usage.", + "stage": "building", + "agent_id": "conv_8d22", + "origin_chat_id": "conv_5b9e", + "autonomy": "propose", + "wake": { + "morning": true, + "related_issues": false + }, + "budget_usd": 100.0, + "created_at": "2026-11-02T13:00:00Z", + "updated_at": "2026-11-02T14:05:00Z", + "started_at": "2026-11-02T14:05:00Z", + "total": 1, + "merged": 0, + "needs_you": 1, + "spend_usd": 2.5 + } + }, + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "", + "stage": "shaping", + "agent_id": "conv_8d22", + "autonomy": "propose", + "wake": { + "morning": false, + "related_issues": false + }, + "created_at": "2026-11-02T14:05:00Z", + "updated_at": "2026-11-02T14:05:00Z", + "total": 0, + "merged": 0, + "needs_you": 0, + "spend_usd": 0.0 + } + } + ], + "error": [ + { + "code": "not_found", + "message": "workstream not found" + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.list.json b/cmd/codeaf/testdata/workstreams-wire/workstream.list.json new file mode 100644 index 0000000000..3a499f0c95 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.list.json @@ -0,0 +1,48 @@ +{ + "method": "workstream.list", + "errors": [ + "busy", + "unknown_method" + ], + "params_type": "WorkstreamListParams", + "result_type": "WorkstreamListResult", + "examples": { + "params": [ + {} + ], + "result": [ + { + "workstreams": [ + { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "stage": "building", + "total": 1, + "merged": 0, + "needs_you": 1, + "spend_usd": 2.5, + "budget_usd": 100.0, + "updated_at": "2026-11-02T14:05:00Z" + } + ] + }, + { + "workstreams": [ + { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "stage": "shaping", + "total": 1, + "updated_at": "2026-11-02T14:05:00Z" + } + ] + } + ], + "error": [ + { + "code": "unknown_method", + "message": "workstreams: not implemented yet" + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.mark_done.json b/cmd/codeaf/testdata/workstreams-wire/workstream.mark_done.json new file mode 100644 index 0000000000..bae070bfbf --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.mark_done.json @@ -0,0 +1,47 @@ +{ + "method": "workstream.mark_done", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamMarkDoneParams", + "result_type": "WorkstreamMarkDoneResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a", + "how_it_went": "Metering shipped.", + "archive_chats": true + } + ], + "result": [ + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "Charge for recorded usage.", + "stage": "done", + "agent_id": "conv_8d22", + "origin_chat_id": "conv_5b9e", + "autonomy": "propose", + "wake": { + "morning": true, + "related_issues": false + }, + "budget_usd": 100.0, + "created_at": "2026-11-02T13:00:00Z", + "updated_at": "2026-11-02T15:00:00Z", + "started_at": "2026-11-02T14:05:00Z", + "total": 1, + "merged": 0, + "needs_you": 1, + "spend_usd": 2.5, + "done_at": "2026-11-02T15:00:00Z", + "how_it_went": "Metering shipped." + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.plan_get.json b/cmd/codeaf/testdata/workstreams-wire/workstream.plan_get.json new file mode 100644 index 0000000000..b5e457560e --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.plan_get.json @@ -0,0 +1,28 @@ +{ + "method": "workstream.plan_get", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPlanGetParams", + "result_type": "WorkstreamPlanGetResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a" + } + ], + "result": [ + { + "plan": { + "workstream_id": "ws_8f2c1a", + "version": 0, + "milestones": [], + "waves": [] + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.plan_set.json b/cmd/codeaf/testdata/workstreams-wire/workstream.plan_set.json new file mode 100644 index 0000000000..ef9aacd6bf --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.plan_set.json @@ -0,0 +1,86 @@ +{ + "method": "workstream.plan_set", + "errors": [ + "bad_params", + "not_found", + "conflict", + "version_conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamPlanSetParams", + "result_type": "WorkstreamPlanSetResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a", + "base_version": 0, + "milestones": [ + { + "id": "ms_usage", + "title": "Recorded usage" + } + ], + "waves": [ + { + "n": 1, + "title": "Metering", + "milestone_id": "ms_usage", + "auto_start": false, + "items": [ + { + "kind": "job", + "factory": "f_local", + "job": 1107 + }, + { + "kind": "draft", + "draft_id": "draft_418c", + "title": "Expose usage totals", + "repo": "acme/billing", + "text": "Show the recorded usage." + } + ] + } + ] + } + ], + "result": [ + { + "plan": { + "workstream_id": "ws_8f2c1a", + "version": 1, + "updated_at": "2026-11-02T14:05:00Z", + "milestones": [ + { + "id": "ms_usage", + "title": "Recorded usage" + } + ], + "waves": [ + { + "n": 1, + "title": "Metering", + "milestone_id": "ms_usage", + "auto_start": false, + "items": [ + { + "kind": "job", + "factory": "f_local", + "job": 1107 + }, + { + "kind": "draft", + "draft_id": "draft_418c", + "title": "Expose usage totals", + "repo": "acme/billing", + "text": "Show the recorded usage." + } + ] + } + ] + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.reopen.json b/cmd/codeaf/testdata/workstreams-wire/workstream.reopen.json new file mode 100644 index 0000000000..8ad474a157 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.reopen.json @@ -0,0 +1,43 @@ +{ + "method": "workstream.reopen", + "errors": [ + "bad_params", + "not_found", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamReopenParams", + "result_type": "WorkstreamReopenResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a" + } + ], + "result": [ + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "Charge for recorded usage.", + "stage": "building", + "agent_id": "conv_8d22", + "origin_chat_id": "conv_5b9e", + "autonomy": "propose", + "wake": { + "morning": true, + "related_issues": false + }, + "budget_usd": 100.0, + "created_at": "2026-11-02T13:00:00Z", + "updated_at": "2026-11-02T15:00:00Z", + "started_at": "2026-11-02T14:05:00Z", + "total": 1, + "merged": 0, + "needs_you": 1, + "spend_usd": 2.5 + } + } + ] + } +} diff --git a/cmd/codeaf/testdata/workstreams-wire/workstream.update.json b/cmd/codeaf/testdata/workstreams-wire/workstream.update.json new file mode 100644 index 0000000000..b9543ee5a1 --- /dev/null +++ b/cmd/codeaf/testdata/workstreams-wire/workstream.update.json @@ -0,0 +1,52 @@ +{ + "method": "workstream.update", + "errors": [ + "bad_params", + "not_found", + "conflict", + "busy", + "unknown_method" + ], + "params_type": "WorkstreamUpdateParams", + "result_type": "WorkstreamUpdateResult", + "examples": { + "params": [ + { + "id": "ws_8f2c1a", + "patch": { + "why": "", + "autonomy": "small", + "wake": { + "morning": false, + "related_issues": true + }, + "budget_usd": 0.0 + } + } + ], + "result": [ + { + "workstream": { + "id": "ws_8f2c1a", + "name": "Usage-based billing", + "why": "", + "stage": "building", + "agent_id": "conv_8d22", + "origin_chat_id": "conv_5b9e", + "autonomy": "small", + "wake": { + "morning": false, + "related_issues": true + }, + "created_at": "2026-11-02T13:00:00Z", + "updated_at": "2026-11-02T14:05:00Z", + "started_at": "2026-11-02T14:05:00Z", + "total": 1, + "merged": 0, + "needs_you": 1, + "spend_usd": 2.5 + } + } + ] + } +} From a1ac718f417fdda00142f2a6f9ad24acb351376f Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 3 Nov 2026 00:35:39 -0500 Subject: [PATCH 2/5] tests: workstream answers through the wire keep the contract's shapes Each method's example params go through the real wire handler against a seeded store; each answer must have its example's keys at every level, the field tables' JSON kinds and enum words, no nulls and integral integers. Every engine-sent error code has a call that checks its code and detail keys. A method, event or error code the contract gains without a case fails. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/appworkstreams_conformance_test.go | 1205 +++++++++++++++++ .../0-workstreams-wire-conformance.md | 7 + 2 files changed, 1212 insertions(+) create mode 100644 cmd/codeaf/appworkstreams_conformance_test.go create mode 100644 docs/changes/unreleased/0-workstreams-wire-conformance.md diff --git a/cmd/codeaf/appworkstreams_conformance_test.go b/cmd/codeaf/appworkstreams_conformance_test.go new file mode 100644 index 0000000000..b09fc9c3c4 --- /dev/null +++ b/cmd/codeaf/appworkstreams_conformance_test.go @@ -0,0 +1,1205 @@ +package main + +import ( + "bufio" + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "math/big" + "os" + "path/filepath" + "reflect" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/appwire" + "github.com/Agent-Field/codeaf/internal/factory/api" + "github.com/Agent-Field/codeaf/internal/workspace" + "github.com/Agent-Field/codeaf/internal/workstream" +) + +type workstreamWireContract struct { + Methods []string `json:"methods"` + Events []struct { + Event string `json:"event"` + DataType string `json:"data_type"` + } `json:"events"` + Errors []workstreamWireErrorType `json:"errors"` + Enums map[string][]string `json:"enums"` + Types map[string]map[string]struct { + Type string `json:"type"` + Required bool `json:"required"` + } `json:"types"` +} + +type workstreamWireErrorType struct { + Code string `json:"code"` + DeskLocal bool `json:"desk_local"` + DetailType *string `json:"detail_type"` +} + +type workstreamWireExamples struct { + Method string `json:"method"` + Event string `json:"event"` + ParamsType string `json:"params_type"` + ResultType string `json:"result_type"` + DataType string `json:"data_type"` + Errors []string `json:"errors"` + Examples struct { + Params []json.RawMessage `json:"params"` + Result []json.RawMessage `json:"result"` + Error []json.RawMessage `json:"error"` + Data []json.RawMessage `json:"data"` + } `json:"examples"` +} + +type workstreamWireDocQuestion struct { + Path string + Question string +} + +// Each entry is a difference between the engine and the wire contract that the +// contract must settle (a doc question), tolerated only at its path. Each +// alternate element must still match its own explicit example and pass +// conforms. An entry is stale once its path no longer differs. +var workstreamWireDocQuestions = []workstreamWireDocQuestion{} + +type workstreamWireComparison struct { + Questions map[string]string + Differed map[string]bool +} + +func workstreamWireDecode(t *testing.T, raw []byte) any { + t.Helper() + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.UseNumber() + var value any + if err := decoder.Decode(&value); err != nil { + t.Fatal(err) + } + var extra any + if err := decoder.Decode(&extra); err != io.EOF { + t.Fatalf("want one JSON value, got trailing data: %v", err) + } + return value +} + +func workstreamWireEncode(t *testing.T, value any) []byte { + t.Helper() + raw, err := json.Marshal(value) + if err != nil { + t.Fatal(err) + } + return raw +} + +func workstreamWireKind(value any) string { + switch value.(type) { + case nil: + return "null" + case map[string]any: + return "object" + case []any: + return "array" + case string: + return "string" + case bool: + return "boolean" + case json.Number: + return "number" + default: + return fmt.Sprintf("%T", value) + } +} + +func workstreamWireKeys(value map[string]any) []string { + keys := make([]string, 0, len(value)) + for key := range value { + keys = append(keys, key) + } + slices.Sort(keys) + return keys +} + +func workstreamWireNoNulls(path string, value any) []string { + var failures []string + switch value := value.(type) { + case nil: + failures = append(failures, path+": null is never allowed; optional fields must be absent") + case map[string]any: + for _, key := range workstreamWireKeys(value) { + failures = append(failures, workstreamWireNoNulls(path+"."+key, value[key])...) + } + case []any: + for i, item := range value { + failures = append(failures, workstreamWireNoNulls(fmt.Sprintf("%s[%d]", path, i), item)...) + } + } + return failures +} + +// conforms is shared by actual wire replies and every declared example type. +// It deliberately uses the independent contract rather than Go wire structs. +func (contract *workstreamWireContract) conforms(path string, value any, typeString string) []string { + if value == nil { + return workstreamWireNoNulls(path, value) + } + want := func() []string { + return []string{fmt.Sprintf("%s: want %s, got %s", path, typeString, workstreamWireKind(value))} + } + if element, array := strings.CutSuffix(typeString, "[]"); array { + items, ok := value.([]any) + if !ok { + return want() + } + var failures []string + for i, item := range items { + failures = append(failures, contract.conforms(fmt.Sprintf("%s[%d]", path, i), item, element)...) + } + return failures + } + if words, enum := contract.Enums[typeString]; enum { + word, ok := value.(string) + if !ok { + return want() + } + if !slices.Contains(words, word) { + return []string{fmt.Sprintf("%s: %q is not in %s %v", path, word, typeString, words)} + } + return nil + } + if fields, object := contract.Types[typeString]; object { + values, ok := value.(map[string]any) + if !ok { + return want() + } + var failures []string + for _, key := range workstreamWireKeys(values) { + field, known := fields[key] + if !known { + failures = append(failures, path+"."+key+": unknown field of "+typeString) + continue + } + failures = append(failures, contract.conforms(path+"."+key, values[key], field.Type)...) + } + var required []string + for key, field := range fields { + if _, present := values[key]; field.Required && !present { + required = append(required, key) + } + } + slices.Sort(required) + for _, key := range required { + failures = append(failures, path+"."+key+": missing required field of "+typeString) + } + return failures + } + switch typeString { + case "string": + if _, ok := value.(string); !ok { + return want() + } + case "string (RFC 3339)": + text, ok := value.(string) + if !ok { + return want() + } + if _, err := time.Parse(time.RFC3339Nano, text); err != nil { + return []string{fmt.Sprintf("%s: invalid RFC 3339 time %q: %v", path, text, err)} + } + case "boolean": + if _, ok := value.(bool); !ok { + return want() + } + case "integer": + number, ok := value.(json.Number) + if !ok { + return want() + } + // Exact arithmetic avoids accepting a rounded fractional number as an integer. + rational, valid := new(big.Rat).SetString(number.String()) + if !valid || !rational.IsInt() || rational.Sign() < 0 { + return []string{fmt.Sprintf("%s: want a non-negative integral JSON number, got %s", path, number)} + } + case "number (USD)": + if _, ok := value.(json.Number); !ok { + return want() + } + case "object": + if _, ok := value.(map[string]any); !ok { + return want() + } + return workstreamWireNoNulls(path, value) + default: + return []string{path + ": unknown contract type " + typeString} + } + return nil +} + +// Prepared params compare arrays by index and length because preparation may +// change only values. Replies compare every array element to the first example; +// an empty example array leaves element validation to the declared field type. +func (comparison *workstreamWireComparison) shape(path string, value, example any, prepared bool) []string { + if value == nil || example == nil { + return []string{path + ": null is never allowed"} + } + if workstreamWireKind(value) != workstreamWireKind(example) { + return []string{fmt.Sprintf("%s: want example kind %s, got %s", path, workstreamWireKind(example), workstreamWireKind(value))} + } + var failures []string + switch example := example.(type) { + case map[string]any: + actual := value.(map[string]any) + for _, key := range workstreamWireKeys(example) { + item, present := actual[key] + if !present { + failures = append(failures, path+"."+key+": example field is absent") + continue + } + failures = append(failures, comparison.shape(path+"."+key, item, example[key], prepared)...) + } + for _, key := range workstreamWireKeys(actual) { + if _, present := example[key]; !present { + failures = append(failures, path+"."+key+": field is absent from example") + } + } + case []any: + actual := value.([]any) + if prepared && len(actual) != len(example) { + failures = append(failures, fmt.Sprintf("%s: preparation changed array length from %d to %d", path, len(example), len(actual))) + } + if len(example) == 0 { + return failures + } + if len(actual) == 0 { + return append(failures, path+": expected a populated example array") + } + for i, item := range actual { + // An answer element is compared with the example element at the + // same index (arrays may mix shapes, such as a job and a draft + // plan item); past the example's end, with its first element. + index := 0 + if i < len(example) { + index = i + } + itemPath := fmt.Sprintf("%s[%d]", path, i) + differences := comparison.shape(itemPath, item, example[index], prepared) + if _, question := comparison.Questions[itemPath]; !prepared && question && i < len(example) && len(differences) > 0 { + comparison.Differed[itemPath] = true + // The question permits only the shape explicitly illustrated for + // this element, so it cannot hide new keys, nulls or wrong kinds. + differences = comparison.shape(itemPath, item, example[i], true) + } + failures = append(failures, differences...) + } + } + return failures +} + +func workstreamWireCheck(t *testing.T, failures []string) { + t.Helper() + for _, failure := range failures { + t.Error(failure) + } +} + +func workstreamWireLoad(t *testing.T) (workstreamWireContract, map[string]workstreamWireExamples) { + t.Helper() + root := filepath.Join("testdata", "workstreams-wire") + raw, err := os.ReadFile(filepath.Join(root, "contract.json")) + if err != nil { + t.Fatal(err) + } + var contract workstreamWireContract + if err := json.Unmarshal(raw, &contract); err != nil { + t.Fatal(err) + } + paths, err := filepath.Glob(filepath.Join(root, "*.json")) + if err != nil { + t.Fatal(err) + } + files := make(map[string]workstreamWireExamples) + for _, path := range paths { + name := strings.TrimSuffix(filepath.Base(path), ".json") + if name == "contract" { + continue + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var examples workstreamWireExamples + if err := json.Unmarshal(raw, &examples); err != nil { + t.Fatalf("%s: %v", path, err) + } + files[name] = examples + } + return contract, files +} + +// Warm would open real provider/process owners. The existing harness already +// supplies their fakes; method dispatch and Close still belong to appBackend. +type workstreamWireBackend struct{ *appBackend } + +func (*workstreamWireBackend) Warm() {} + +type workstreamWireUnavailable struct{ *appBackend } + +func (*workstreamWireUnavailable) Warm() {} +func (*workstreamWireUnavailable) ServesWorkstreams() bool { return false } + +type workstreamWireConnection struct { + In *io.PipeWriter + Lines <-chan []byte + Err <-chan error + Next int +} + +func workstreamWireConnect(t *testing.T, backend appwire.Engine) *workstreamWireConnection { + t.Helper() + inR, inW := io.Pipe() + outR, outW := io.Pipe() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + server := appwire.NewServer(outW) + go func() { + done <- server.Serve(ctx, inR, appwire.Hello{Version: "test"}, backend) + _ = outW.Close() + }() + lines := make(chan []byte, 32) + readErr := make(chan error, 1) + readerDone := make(chan struct{}) + go func() { + defer close(readerDone) + defer close(lines) + scanner := bufio.NewScanner(outR) + scanner.Buffer(make([]byte, 64<<10), 4<<20) + for scanner.Scan() { + select { + case lines <- bytes.Clone(scanner.Bytes()): + case <-ctx.Done(): + return + } + } + readErr <- scanner.Err() + }() + t.Cleanup(func() { + _ = inW.Close() + select { + case err := <-done: + if err != nil { + t.Errorf("wire Serve: %v", err) + } + case <-time.After(30 * time.Second): + t.Error("wire Serve did not close the backend") + } + cancel() + _ = inR.Close() + _ = outR.Close() + select { + case <-readerDone: + case <-time.After(5 * time.Second): + t.Error("wire reader did not stop") + } + }) + connection := &workstreamWireConnection{In: inW, Lines: lines, Err: readErr} + hello := connection.read(t) + if hello["event"] != appwire.EventHello { + t.Fatalf("want hello before replies, got %v", hello) + } + return connection +} + +func (connection *workstreamWireConnection) read(t *testing.T) map[string]any { + t.Helper() + select { + case raw, open := <-connection.Lines: + if !open { + t.Fatalf("wire ended before reply: %v", <-connection.Err) + } + frame, ok := workstreamWireDecode(t, raw).(map[string]any) + if !ok { + t.Fatalf("wire frame is not an object: %s", raw) + } + return frame + case <-time.After(30 * time.Second): + t.Fatal("wire reply timed out") + return nil + } +} + +func (connection *workstreamWireConnection) call(t *testing.T, method string, params any) map[string]any { + t.Helper() + connection.Next++ + raw := workstreamWireEncode(t, map[string]any{"id": connection.Next, "method": method, "params": params}) + if _, err := connection.In.Write(append(raw, '\n')); err != nil { + t.Fatal(err) + } + for { + frame := connection.read(t) + id, reply := frame["id"].(json.Number) + if !reply { + if _, event := frame["event"].(string); !event { + t.Fatalf("frame has neither a numeric id nor an event: %v", frame) + } + continue + } + if id.String() != strconv.Itoa(connection.Next) { + t.Fatalf("reply id %s, want %d", id, connection.Next) + } + if len(frame) != 2 { + t.Fatalf("reply must have id and one result or error, got keys %v", workstreamWireKeys(frame)) + } + return frame + } +} + +func workstreamWireResult(t *testing.T, frame map[string]any) any { + t.Helper() + result, present := frame["result"] + if !present || result == nil { + t.Fatalf("want result without error, got %v", frame) + } + return result +} + +type workstreamWireFixture struct { + Backend *appBackend + Recorder *appRecorder + IDs map[string]string + Now time.Time + Stream appwire.Workstream +} + +func workstreamWireSeed(t *testing.T, name string, situation int) *workstreamWireFixture { + t.Helper() + backend, recorder, _, origin := workstreamBackend(t) + now := time.Now().UTC().Truncate(time.Millisecond) + backend.now = func() time.Time { return now } + backend.workstreams.archiveChats = func([]string) error { return nil } + fixture := &workstreamWireFixture{ + Backend: backend, Recorder: recorder, Now: now, + IDs: map[string]string{"conv_5b9e": origin.ID, "ms_usage": "conformance-milestone", "draft_418c": "conformance-draft"}, + } + if name == "workstream.create" { + // An unobserved member job keeps all optional numbers unknown on create. + return fixture + } + zero := name == "workstream.get" && situation == 1 + unknown := name == "workstream.list" && situation == 1 + params := appwire.WorkstreamCreateParams{Name: "Usage-based billing", Why: workstreamPtr("Charge for recorded usage.")} + if !zero { + params.OriginChatID = &origin.ID + if name != "collections.add" { + ref := workstreamJobRef("1107") + ref.Factory = workstreamPtr("f_local") + params.Refs = []appwire.WorkstreamRef{ref} + } + } + if zero { + params.Why = nil + } + made, err := backend.WorkstreamCreate(params) + if err != nil { + t.Fatal(err) + } + fixture.Stream = made.Workstream + fixture.IDs["ws_8f2c1a"] = made.Workstream.ID + fixture.IDs["conv_8d22"] = made.Workstream.AgentID + if unknown { + backend.workstreams.spend = func(string) *float64 { return nil } + } else { + backend.workstreams.queue = func(context.Context, string) (api.Queue, error) { + return api.Queue{Jobs: []api.Job{{ID: 1107, Num: 2107, State: "needs_you", Cost: 2.5}}}, nil + } + } + if !zero && !unknown && strings.HasPrefix(name, "workstream.") && name != "workstream.changed" { + if _, err := backend.WorkstreamUpdate(appwire.WorkstreamUpdateParams{ID: made.Workstream.ID, Patch: appwire.WorkstreamPatch{BudgetUSD: workstreamPtr(100.0)}}); err != nil { + t.Fatal(err) + } + } + pages, err := backend.WorkstreamPageList(appwire.WorkstreamPageListParams{WorkstreamID: made.Workstream.ID}) + if err != nil || len(pages.Pages) != 1 { + t.Fatalf("seed Overview: %+v, %v", pages, err) + } + fixture.IDs["pg_31d0"] = pages.Pages[0].ID + switch name { + case "workstream.reopen": + if _, err := backend.WorkstreamMarkDone(appwire.WorkstreamMarkDoneParams{ID: made.Workstream.ID}); err != nil { + t.Fatal(err) + } + case "page.update", "page.delete", "page.move", "workstream.changed": + parent := fixture.IDs["pg_31d0"] + if name == "page.move" { + parent = "" + } + fixture.page(t, "pg_32e1", parent) + if name == "page.move" && situation == 1 { + fixture.page(t, "pg_548c", fixture.IDs["pg_31d0"]) + } + if name == "page.delete" { + fixture.page(t, "conformance-descendant", fixture.IDs["pg_32e1"]) + } + case "page.history", "page.restore": + fixture.saveByAgent(t) + } + if name == "record.list" || name == "record.keep" || name == "record.drop" || name == "record.supersede" || name == "workstream.changed" { + fixture.record(t) + if name == "record.supersede" { + if _, err := backend.WorkstreamRecordKeep(appwire.WorkstreamRecordKeepParams{ID: fixture.IDs["rec_7a1e"]}); err != nil { + t.Fatal(err) + } + } + } + if strings.HasPrefix(name, "goahead.") || name == "workstream.changed" { + fixture.plan(t) + if name != "goahead.grant" { + fixture.grant(t) + } + } + return fixture +} + +func (fixture *workstreamWireFixture) page(t *testing.T, alias, parent string) { + t.Helper() + params := appwire.WorkstreamPageCreateParams{WorkstreamID: fixture.Stream.ID, Title: "Metering decisions"} + if parent != "" { + params.ParentID = &parent + } + page, err := fixture.Backend.WorkstreamPageCreate(params) + if err != nil { + t.Fatal(err) + } + fixture.IDs[alias] = page.Page.ID +} + +func (fixture *workstreamWireFixture) source() *appwire.WorkstreamSource { + return &appwire.WorkstreamSource{Kind: "message", ID: "msg_19a2", ChatID: workstreamPtr(fixture.IDs["conv_5b9e"]), Label: workstreamPtr("Billing scope")} +} + +func (fixture *workstreamWireFixture) saveByAgent(t *testing.T) { + t.Helper() + store, err := fixture.Backend.workstreamStore(false) + if err != nil || store == nil { + t.Fatalf("seed page store: %v", err) + } + _, _, err = store.SavePage(context.Background(), fixture.IDs["pg_31d0"], 1, + json.RawMessage(`{"type":"doc","content":[]}`), workstream.Attribution{By: "agent", From: workstreamSource(fixture.source())}) + if err != nil { + t.Fatal(err) + } +} + +func (fixture *workstreamWireFixture) record(t *testing.T) { + t.Helper() + record, err := fixture.Backend.WorkstreamRecordPropose(appwire.WorkstreamRecordProposeParams{ + WorkstreamID: fixture.Stream.ID, Kind: "question", Text: "Should retries count toward usage?", Source: fixture.source(), + }) + if err != nil { + t.Fatal(err) + } + fixture.IDs["rec_7a1e"] = record.Record.ID +} + +func (fixture *workstreamWireFixture) plan(t *testing.T) { + t.Helper() + _, err := fixture.Backend.WorkstreamPlanSet(appwire.WorkstreamPlanSetParams{ + ID: fixture.Stream.ID, + Milestones: []appwire.WorkstreamMilestone{{ID: fixture.IDs["ms_usage"], Title: "Recorded usage"}}, + Waves: []appwire.WorkstreamWave{{N: 1, Title: workstreamPtr("Metering"), MilestoneID: workstreamPtr(fixture.IDs["ms_usage"]), Items: []appwire.WorkstreamPlanItem{ + {Kind: "job", Factory: workstreamPtr("f_local"), Job: workstreamPtr(1107)}, + {Kind: "draft", DraftID: workstreamPtr(fixture.IDs["draft_418c"]), Title: workstreamPtr("Expose usage totals"), Repo: workstreamPtr("acme/billing"), Text: workstreamPtr("Show the recorded usage.")}, + }}}, + }) + if err != nil { + t.Fatal(err) + } +} + +func (fixture *workstreamWireFixture) grant(t *testing.T) { + t.Helper() + grant, err := fixture.Backend.WorkstreamGoAheadGrant(appwire.WorkstreamGoAheadGrantParams{ + WorkstreamID: fixture.Stream.ID, Scope: "job", Factory: workstreamPtr("f_local"), Job: workstreamPtr(1107), CapUSD: 40, EndsAt: fixture.Now.Add(time.Hour), + }) + if err != nil { + t.Fatal(err) + } + fixture.IDs["ga_0c44"] = grant.GoAhead.ID +} + +// The decoded example is a fresh tree on every call. Only existing values are +// replaced; shape(..., true) independently proves that keys and kinds survive. +func (fixture *workstreamWireFixture) prepare(t *testing.T, raw json.RawMessage) map[string]any { + t.Helper() + var replace func(any) any + replace = func(value any) any { + switch value := value.(type) { + case map[string]any: + for key, item := range value { + if key == "ends_at" { + value[key] = fixture.Now.Add(time.Hour).Format(time.RFC3339Nano) + } else { + value[key] = replace(item) + } + } + case []any: + for i, item := range value { + value[i] = replace(item) + } + case string: + if seeded, known := fixture.IDs[value]; known { + return seeded + } + } + return value + } + return replace(workstreamWireDecode(t, raw)).(map[string]any) +} + +func (contract *workstreamWireContract) errorType(code string) (workstreamWireErrorType, bool) { + for _, definition := range contract.Errors { + if definition.Code == code { + return definition, true + } + } + return workstreamWireErrorType{}, false +} + +func (contract *workstreamWireContract) checkError(t *testing.T, path string, value any, code string, example any, exactMessage bool) { + t.Helper() + object, ok := value.(map[string]any) + if !ok { + t.Fatalf("%s: want error object, got %s", path, workstreamWireKind(value)) + } + workstreamWireCheck(t, workstreamWireNoNulls(path, object)) + if object["code"] != code { + t.Errorf("%s.code: got %v, want %s", path, object["code"], code) + } + message, ok := object["message"].(string) + if !ok || strings.TrimSpace(message) == "" { + t.Errorf("%s.message: want non-empty string, got %v", path, object["message"]) + } + definition, known := contract.errorType(code) + if !known || definition.DeskLocal { + t.Fatalf("%s: %s is not an engine-sent contract error", path, code) + } + for key := range object { + if key != "code" && key != "message" && key != "detail" { + t.Errorf("%s.%s: unexpected error field", path, key) + } + } + detail, present := object["detail"] + if definition.DetailType == nil { + if present { + t.Errorf("%s.detail: must be absent for %s", path, code) + } + } else if !present { + t.Errorf("%s.detail: missing %s", path, *definition.DetailType) + } else { + workstreamWireCheck(t, contract.conforms(path+".detail", detail, *definition.DetailType)) + } + if example != nil { + comparison := &workstreamWireComparison{} + workstreamWireCheck(t, comparison.shape(path, object, example, false)) + if exactMessage && object["message"] != example.(map[string]any)["message"] { + t.Errorf("%s.message: got %q, want %q", path, message, example.(map[string]any)["message"]) + } + } +} + +type workstreamWireErrorCase struct { + Name string + Method string + Code string + Example string +} + +// TestAppWorkstreamWireConformance keeps the table independent of the fixtures: +// adding a contract method, event, file or engine-sent error must require a case. +func TestAppWorkstreamWireConformance(t *testing.T) { + contract, files := workstreamWireLoad(t) + cases := []string{ + "workstream.list", "workstream.get", "workstream.create", "workstream.update", + "workstream.mark_done", "workstream.reopen", "workstream.plan_get", "workstream.plan_set", + "page.list", "page.get", "page.save", "page.create", "page.update", "page.move", + "page.delete", "page.history", "page.restore", "record.list", "record.propose", + "record.keep", "record.drop", "record.supersede", "goahead.list", "goahead.grant", + "goahead.revoke", "collections.add", "workstream.changed", + } + errorCases := []workstreamWireErrorCase{ + {Name: "bad_params", Method: "workstream.get", Code: "bad_params"}, + {Name: "not_found", Method: "workstream.get", Code: "not_found", Example: "workstream.get"}, + {Name: "conflict", Method: "workstream.create", Code: "conflict"}, + {Name: "collection_owner", Method: "collections.add", Code: "in_other_workstream", Example: "collections.add"}, + {Name: "create_owner", Method: "workstream.create", Code: "in_other_workstream", Example: "collections.add"}, + {Name: "page_version", Method: "page.save", Code: "version_conflict", Example: "page.save"}, + {Name: "plan_version", Method: "workstream.plan_set", Code: "version_conflict", Example: "page.save"}, + {Name: "cycle", Method: "page.move", Code: "cycle"}, + {Name: "too_large", Method: "page.save", Code: "too_large"}, + {Name: "unknown_method", Method: "workstream.list", Code: "unknown_method", Example: "workstream.list"}, + {Name: "busy", Method: "workstream.update", Code: "busy"}, + } + comparison := &workstreamWireComparison{Questions: make(map[string]string), Differed: make(map[string]bool)} + for _, question := range workstreamWireDocQuestions { + if question.Path == "" || strings.TrimSpace(question.Question) == "" || comparison.Questions[question.Path] != "" { + t.Fatalf("invalid or duplicate doc question: %+v", question) + } + comparison.Questions[question.Path] = question.Question + } + + // (a), (c): coverage is checked in both directions, including auxiliary files + // and all engine-sent errors. No busy exemption is needed: it has a real case. + if !t.Run("coverage", func(t *testing.T) { + workstreamWireCoverage(t, &contract, files, cases, errorCases) + }) { + return + } + + // (b), (d): the same strict type validator checks all documentation examples. + t.Run("shape_rules", func(t *testing.T) { + for _, rule := range []struct { + Name string + Type string + JSON string + Valid bool + }{ + {"string", "string", `"text"`, true}, + {"string_number", "string", `1`, false}, + {"nano_time", "string (RFC 3339)", `"2026-11-02T14:05:00.123456789Z"`, true}, + {"bad_time", "string (RFC 3339)", `"yesterday"`, false}, + {"boolean", "boolean", `false`, true}, + {"boolean_string", "boolean", `"false"`, false}, + {"zero", "integer", `0`, true}, + {"integral_decimal", "integer", `1.0`, true}, + {"fraction", "integer", `1.5`, false}, + {"precise_fraction", "integer", `9007199254740992.1`, false}, + {"negative", "integer", `-1`, false}, + {"numeric_string", "integer", `"1"`, false}, + {"usd", "number (USD)", `2.5`, true}, + {"usd_string", "number (USD)", `"2.5"`, false}, + {"enum", "WorkstreamStage", `"building"`, true}, + {"unknown_enum", "WorkstreamStage", `"working"`, false}, + {"empty_array", "WorkstreamChangedPart[]", `[]`, true}, + {"unknown_array_enum", "WorkstreamChangedPart[]", `["jobs","other"]`, false}, + {"null_array", "WorkstreamChangedPart[]", `null`, false}, + {"optional_absent", "WorkstreamChanged", `{"id":"seeded"}`, true}, + {"optional_null", "WorkstreamChanged", `{"id":"seeded","parts":null}`, false}, + {"missing_required", "WorkstreamChanged", `{"parts":[]}`, false}, + {"unknown_field", "WorkstreamChanged", `{"id":"seeded","extra":true}`, false}, + {"opaque_object", "object", `{"arbitrary":[{"value":0}]}`, true}, + {"opaque_nested_null", "object", `{"arbitrary":[null]}`, false}, + } { + t.Run(rule.Name, func(t *testing.T) { + path := "shape rule " + rule.Name + " " + failures := contract.conforms(path, workstreamWireDecode(t, []byte(rule.JSON)), rule.Type) + if (len(failures) == 0) != rule.Valid { + t.Errorf("valid = %v, want %v: %v", len(failures) == 0, rule.Valid, failures) + } + for _, failure := range failures { + if !strings.HasPrefix(failure, path) { + t.Errorf("failure lost its JSON path: %s", failure) + } + } + }) + } + }) + + // (b), (d): malformed examples are contract failures, before any engine call. + t.Run("doc_self_check", func(t *testing.T) { + for _, name := range cases { + t.Run(name, func(t *testing.T) { + examples := files[name] + for _, part := range []struct { + Name string + Type string + Rows []json.RawMessage + }{ + {"params", examples.ParamsType, examples.Examples.Params}, + {"result", examples.ResultType, examples.Examples.Result}, + {"data", examples.DataType, examples.Examples.Data}, + } { + for i, raw := range part.Rows { + typeString := part.Type + if name == "collections.add" { + typeString = "object" + } + path := fmt.Sprintf("%s %s[%d] ", name, part.Name, i) + workstreamWireCheck(t, contract.conforms(path, workstreamWireDecode(t, raw), typeString)) + } + } + for i, raw := range examples.Examples.Error { + value := workstreamWireDecode(t, raw) + object, ok := value.(map[string]any) + if !ok { + t.Fatalf("%s error[%d] is not an object", name, i) + } + code, ok := object["code"].(string) + if !ok || (name != "collections.add" && !slices.Contains(examples.Errors, code)) { + t.Fatalf("%s error[%d] has unlisted code %v", name, i, object["code"]) + } + contract.checkError(t, fmt.Sprintf("%s error[%d]", name, i), value, code, nil, false) + } + }) + } + }) + + // (b), (d): each params/result pairing gets an isolated seeded situation. + // A single params example with multiple results exercises every situation. + for _, name := range cases { + t.Run(name, func(t *testing.T) { + examples := files[name] + if examples.Event != "" { + workstreamWireEvent(t, &contract, examples, comparison) + return + } + count := max(len(examples.Examples.Params), len(examples.Examples.Result)) + for i := 0; i < count; i++ { + t.Run(fmt.Sprintf("example_%d", i), func(t *testing.T) { + paramsIndex, resultIndex := i, i + if len(examples.Examples.Params) == 1 { + paramsIndex = 0 + } + if len(examples.Examples.Result) == 1 { + resultIndex = 0 + } + fixture := workstreamWireSeed(t, name, i) + params := fixture.prepare(t, examples.Examples.Params[paramsIndex]) + paramsPath := fmt.Sprintf("%s params[%d] ", name, paramsIndex) + workstreamWireCheck(t, comparison.shape(paramsPath, params, workstreamWireDecode(t, examples.Examples.Params[paramsIndex]), true)) + if examples.ParamsType != "" { + workstreamWireCheck(t, contract.conforms(paramsPath, params, examples.ParamsType)) + } + connection := workstreamWireConnect(t, &workstreamWireBackend{appBackend: fixture.Backend}) + result := workstreamWireResult(t, connection.call(t, name, params)) + resultPath := fmt.Sprintf("%s result[%d] ", name, resultIndex) + if examples.ResultType != "" { + workstreamWireCheck(t, contract.conforms(resultPath, result, examples.ResultType)) + } + workstreamWireCheck(t, workstreamWireNoNulls(resultPath, result)) + workstreamWireCheck(t, comparison.shape(resultPath, result, workstreamWireDecode(t, examples.Examples.Result[resultIndex]), false)) + if name == "collections.add" { + workstreamWireCollectionResult(t, result) + } + }) + } + }) + } + + // (c): every engine-sent code is produced by a real wire call, including + // both ownership doors and both optimistic-version doors. + t.Run("errors", func(t *testing.T) { + for _, errorCase := range errorCases { + t.Run(errorCase.Name, func(t *testing.T) { + workstreamWireError(t, &contract, files, errorCase) + }) + } + }) + + // (d): fixing an example or changing the observed discrepancy requires + // removing its exception; doc questions cannot silently become permanent. + t.Run("doc_questions_are_current", func(t *testing.T) { + for _, question := range workstreamWireDocQuestions { + if !comparison.Differed[question.Path] { + t.Errorf("stale doc question at %s: %s", question.Path, question.Question) + } + } + }) +} + +func workstreamWireCoverage(t *testing.T, contract *workstreamWireContract, files map[string]workstreamWireExamples, cases []string, errorCases []workstreamWireErrorCase) { + t.Helper() + if !reflect.DeepEqual(contract.Methods, appwire.WorkstreamMethods) { + t.Errorf("contract methods/order %v differ from appwire.WorkstreamMethods %v", contract.Methods, appwire.WorkstreamMethods) + } + counts := make(map[string]int) + for _, name := range cases { + counts[name]++ + if _, exists := files[name]; !exists { + t.Errorf("stale case %s has no example file", name) + } + } + events := make(map[string]string) + for _, event := range contract.Events { + if _, duplicate := events[event.Event]; duplicate { + t.Errorf("duplicate contract event %s", event.Event) + } + events[event.Event] = event.DataType + if _, exists := files[event.Event]; !exists || counts[event.Event] != 1 { + t.Errorf("event %s needs one file and one case", event.Event) + } + } + methods := make(map[string]bool) + for _, name := range contract.Methods { + if methods[name] { + t.Errorf("duplicate contract method %s", name) + } + methods[name] = true + if _, exists := files[name]; !exists || counts[name] != 1 { + t.Errorf("method %s needs one file and one case, got %d cases", name, counts[name]) + } + } + for name, examples := range files { + if counts[name] != 1 { + t.Errorf("file %s.json needs exactly one case, got %d", name, counts[name]) + } + if dataType, event := events[name]; event { + if examples.Event != name || examples.Method != "" || examples.DataType != dataType || len(examples.Examples.Data) == 0 { + t.Errorf("event file %s disagrees with contract or lacks examples", name) + } + continue + } + if !methods[name] && name != "collections.add" { + t.Errorf("file/case %s names no contract method or event", name) + } + if examples.Method != name || examples.Event != "" || len(examples.Examples.Params) == 0 || len(examples.Examples.Result) == 0 { + t.Errorf("method file %s disagrees with its name or lacks examples", name) + } + params, results := len(examples.Examples.Params), len(examples.Examples.Result) + if params != results && params != 1 && results != 1 { + t.Errorf("%s: unpaired example counts %d/%d need explicit situations", name, params, results) + } + if name != "collections.add" { + for _, typeString := range []string{examples.ParamsType, examples.ResultType} { + if _, known := contract.Types[typeString]; !known { + t.Errorf("%s: missing declared type %q", name, typeString) + } + } + } + for _, code := range examples.Errors { + if _, known := contract.errorType(code); !known { + t.Errorf("%s: unlisted error code %s", name, code) + } + } + } + if !slices.Contains(appwire.Methods(), "collections.add") { + t.Error("collections.add is not a registered wire method") + } + coveredErrors := make(map[string]bool) + caseNames := make(map[string]bool) + for _, errorCase := range errorCases { + if caseNames[errorCase.Name] { + t.Errorf("duplicate error case %s", errorCase.Name) + } + caseNames[errorCase.Name] = true + definition, known := contract.errorType(errorCase.Code) + if !known || definition.DeskLocal { + t.Errorf("stale error case %s: code %s is not engine-sent", errorCase.Name, errorCase.Code) + } + if _, exists := files[errorCase.Method]; !exists { + t.Errorf("stale error case %s: missing method file %s", errorCase.Name, errorCase.Method) + } + if errorCase.Method != "collections.add" && !slices.Contains(files[errorCase.Method].Errors, errorCase.Code) { + t.Errorf("%s does not list %s", errorCase.Method, errorCase.Code) + } + if errorCase.Example != "" { + found := false + for _, raw := range files[errorCase.Example].Examples.Error { + if workstreamWireDecode(t, raw).(map[string]any)["code"] == errorCase.Code { + found = true + } + } + if !found { + t.Errorf("stale error example %s/%s", errorCase.Example, errorCase.Code) + } + } + coveredErrors[errorCase.Code] = true + } + listedErrors := make(map[string]bool) + for _, definition := range contract.Errors { + if listedErrors[definition.Code] { + t.Errorf("duplicate contract error %s", definition.Code) + } + listedErrors[definition.Code] = true + if !definition.DeskLocal && !coveredErrors[definition.Code] { + t.Errorf("engine-sent contract error %s has no wire case", definition.Code) + } + } +} + +func workstreamWireCollectionResult(t *testing.T, result any) { + t.Helper() + object := result.(map[string]any) + added, addedArray := object["added"].([]any) + already, alreadyArray := object["already"].([]any) + if !addedArray || len(added) != 1 || !alreadyArray || len(already) != 0 { + t.Fatalf("collections.add result: want one added job and empty already array, got %v", result) + } + for i, value := range added { + member := value.(map[string]any) + if member["kind"] != "factory_job" || member["id"] != "1107" || member["factory"] != "f_local" { + t.Errorf("collections.add .added[%d]: wrong factory job ref %v", i, member) + } + at, ok := member["added_at"].(string) + if _, err := time.Parse(time.RFC3339Nano, at); !ok || err != nil { + t.Errorf("collections.add .added[%d].added_at: want RFC 3339 string, got %v", i, member["added_at"]) + } + } +} + +func workstreamWireEvent(t *testing.T, contract *workstreamWireContract, examples workstreamWireExamples, comparison *workstreamWireComparison) { + t.Helper() + fixture := workstreamWireSeed(t, "workstream.changed", 0) + before := len(workstreamEvents(fixture.Recorder)) + connection := workstreamWireConnect(t, &workstreamWireBackend{appBackend: fixture.Backend}) + workstreamWireResult(t, connection.call(t, "workstream.update", map[string]any{ + "id": fixture.Stream.ID, "patch": map[string]any{"why": "Updated through the wire"}, + })) + events := workstreamEvents(fixture.Recorder) + if len(events) <= before { + t.Fatal("wire mutation did not emit workstream.changed") + } + for i, event := range events { + value := workstreamWireDecode(t, workstreamWireEncode(t, event)) + path := fmt.Sprintf("workstream.changed data[%d] ", i) + workstreamWireCheck(t, contract.conforms(path, value, examples.DataType)) + exampleIndex := 0 + if _, parts := value.(map[string]any)["parts"]; !parts { + exampleIndex = 1 + } + if exampleIndex >= len(examples.Examples.Data) { + t.Fatalf("%s: no example for parts presence", path) + } + workstreamWireCheck(t, comparison.shape(path, value, workstreamWireDecode(t, examples.Examples.Data[exampleIndex]), false)) + if event.ID != fixture.Stream.ID { + t.Errorf("%s.id: got %s, want seeded workstream %s", path, event.ID, fixture.Stream.ID) + } + } +} + +func workstreamWireError(t *testing.T, contract *workstreamWireContract, files map[string]workstreamWireExamples, errorCase workstreamWireErrorCase) { + t.Helper() + fixture := workstreamWireSeed(t, "workstream.get", 0) + backend := fixture.Backend + examples := files[errorCase.Method] + params := fixture.prepare(t, examples.Examples.Params[0]) + var engine appwire.Engine = &workstreamWireBackend{appBackend: backend} + release := func() {} + switch errorCase.Name { + case "bad_params": + delete(params, "id") + case "not_found": + params["id"] = "missing-workstream" + case "conflict": + params["name"] = fixture.Stream.Name + case "collection_owner": + other := workstreamCreateForTest(t, backend, "Account migration") + params["id"] = other.ID + case "create_owner": + params["name"] = "Another workstream" + case "page_version": + fixture.saveByAgent(t) + case "plan_version": + fixture.plan(t) + case "cycle": + fixture.page(t, "pg_32e1", fixture.IDs["pg_31d0"]) + fixture.page(t, "conformance-descendant", fixture.IDs["pg_32e1"]) + params["id"] = fixture.IDs["pg_32e1"] + params["parent_id"] = fixture.IDs["conformance-descendant"] + case "too_large": + params["doc"] = map[string]any{"type": "doc", "content": []any{ + map[string]any{"type": "paragraph", "content": []any{ + map[string]any{"type": "text", "text": strings.Repeat("x", (1<<20)+1)}, + }}, + }} + case "unknown_method": + engine = &workstreamWireUnavailable{appBackend: backend} + case "busy": + release = workstreamWireHoldWriter(t, backend) + default: + t.Fatalf("unimplemented error case %s", errorCase.Name) + } + if errorCase.Name != "bad_params" && errorCase.Name != "too_large" { + comparison := &workstreamWireComparison{} + workstreamWireCheck(t, comparison.shape(errorCase.Method+" error params ", params, workstreamWireDecode(t, examples.Examples.Params[0]), true)) + } + connection := workstreamWireConnect(t, engine) + frame := connection.call(t, errorCase.Method, params) + release() + value, present := frame["error"] + if !present { + t.Fatalf("%s: want %s error, got %v", errorCase.Method, errorCase.Code, frame) + } + var example any + if errorCase.Example != "" { + for _, raw := range files[errorCase.Example].Examples.Error { + candidate := workstreamWireDecode(t, raw).(map[string]any) + if candidate["code"] == errorCase.Code { + example = candidate + break + } + } + } + contract.checkError(t, errorCase.Method+" error", value, errorCase.Code, example, errorCase.Code == "unknown_method") + object := value.(map[string]any) + if errorCase.Code == "in_other_workstream" { + detail, ok := object["detail"].(map[string]any) + if !ok || detail["workstream_id"] != fixture.Stream.ID || detail["workstream_name"] != fixture.Stream.Name || detail["factory"] != "f_local" || detail["job"] != json.Number("1107") { + t.Errorf("ownership detail does not identify the seeded owner/job: %v", object["detail"]) + } + } + if errorCase.Code == "version_conflict" { + detail, ok := object["detail"].(map[string]any) + id, version, by := fixture.IDs["pg_31d0"], "2", "agent" + if errorCase.Name == "plan_version" { + id, version, by = fixture.Stream.ID, "1", "you" + } + if !ok || detail["id"] != id || detail["version"] != json.Number(version) || detail["updated_by"] != by { + t.Errorf("stale detail does not describe the current seeded version: %v", object["detail"]) + } + } +} + +// Hold a writer on a separate handle to the same real store. The backend must +// exhaust the normal busy timeout and translate that refusal on the wire. +func workstreamWireHoldWriter(t *testing.T, backend *appBackend) func() { + t.Helper() + other, err := workspace.Open(backend.collectionsPath) + if err != nil { + t.Fatal(err) + } + entered := make(chan struct{}) + unlock := make(chan struct{}) + done := make(chan error, 1) + go func() { + done <- other.Write(context.Background(), func(*sql.Tx) error { + close(entered) + <-unlock + return nil + }) + }() + var once sync.Once + release := func() { + once.Do(func() { + close(unlock) + select { + case err := <-done: + if err != nil { + t.Errorf("held writer: %v", err) + } + case <-time.After(30 * time.Second): + t.Error("held writer did not release") + } + if err := other.Close(); err != nil { + t.Errorf("held writer close: %v", err) + } + }) + } + t.Cleanup(release) + select { + case <-entered: + case err := <-done: + // Put the result back so the shared cleanup can still consume it. + done <- err + t.Fatalf("writer did not take the lock: %v", err) + case <-time.After(30 * time.Second): + t.Fatal("writer did not take the lock") + } + return release +} diff --git a/docs/changes/unreleased/0-workstreams-wire-conformance.md b/docs/changes/unreleased/0-workstreams-wire-conformance.md new file mode 100644 index 0000000000..6df7b4e217 --- /dev/null +++ b/docs/changes/unreleased/0-workstreams-wire-conformance.md @@ -0,0 +1,7 @@ +--- +kind: internal +title: Workstream methods are checked against the wire contract's examples +pr: 0 +surface: [app] +invalidates: [] +--- From 10fce79b396822a60e07846b3e23627c56958985 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 3 Nov 2026 00:35:58 -0500 Subject: [PATCH 3/5] docs: change entry for the workstream wire conformance test Co-Authored-By: Claude Opus 5.5 --- ...s-wire-conformance.md => 2288-workstreams-wire-conformance.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/changes/unreleased/{0-workstreams-wire-conformance.md => 2288-workstreams-wire-conformance.md} (100%) diff --git a/docs/changes/unreleased/0-workstreams-wire-conformance.md b/docs/changes/unreleased/2288-workstreams-wire-conformance.md similarity index 100% rename from docs/changes/unreleased/0-workstreams-wire-conformance.md rename to docs/changes/unreleased/2288-workstreams-wire-conformance.md From fbf26899dbc5d9f4f21641415dd5e5feffa3464a Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 3 Nov 2026 00:57:01 -0500 Subject: [PATCH 4/5] docs: the conformance change entry names its pull request Co-Authored-By: Claude Opus 5.5 --- docs/changes/unreleased/2288-workstreams-wire-conformance.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/changes/unreleased/2288-workstreams-wire-conformance.md b/docs/changes/unreleased/2288-workstreams-wire-conformance.md index 6df7b4e217..1aad7e6fc0 100644 --- a/docs/changes/unreleased/2288-workstreams-wire-conformance.md +++ b/docs/changes/unreleased/2288-workstreams-wire-conformance.md @@ -1,7 +1,7 @@ --- kind: internal title: Workstream methods are checked against the wire contract's examples -pr: 0 +pr: 2288 surface: [app] invalidates: [] --- From 58e66da8bb7388bbc7dccdc24f89b0b0d2eedd5b Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 3 Nov 2026 01:00:49 -0500 Subject: [PATCH 5/5] tests: conformance replies carry every illustrated array element A reply array shorter than its example no longer skips the elements the example illustrates (a plan's draft item, an unattributed page version); page.delete must return exactly the seeded parent and descendant. Request writes to the wire are bounded like replies. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/appworkstreams_conformance_test.go | 125 ++++++++++++++---- 1 file changed, 101 insertions(+), 24 deletions(-) diff --git a/cmd/codeaf/appworkstreams_conformance_test.go b/cmd/codeaf/appworkstreams_conformance_test.go index b09fc9c3c4..f39b158764 100644 --- a/cmd/codeaf/appworkstreams_conformance_test.go +++ b/cmd/codeaf/appworkstreams_conformance_test.go @@ -76,6 +76,14 @@ type workstreamWireComparison struct { Differed map[string]bool } +type workstreamWireShapeMode int + +const ( + workstreamWirePreparedShape workstreamWireShapeMode = iota + workstreamWireReplyShape + workstreamWireEventShape +) + func workstreamWireDecode(t *testing.T, raw []byte) any { t.Helper() decoder := json.NewDecoder(bytes.NewReader(raw)) @@ -244,9 +252,11 @@ func (contract *workstreamWireContract) conforms(path string, value any, typeStr } // Prepared params compare arrays by index and length because preparation may -// change only values. Replies compare every array element to the first example; -// an empty example array leaves element validation to the declared field type. -func (comparison *workstreamWireComparison) shape(path string, value, example any, prepared bool) []string { +// change only values. Replies require every illustrated index and compare each +// element with the same example index, falling back to the first past its end. +// Events may list fewer parts. An empty example array leaves element validation +// to the declared field type. +func (comparison *workstreamWireComparison) shape(path string, value, example any, mode workstreamWireShapeMode) []string { if value == nil || example == nil { return []string{path + ": null is never allowed"} } @@ -263,7 +273,7 @@ func (comparison *workstreamWireComparison) shape(path string, value, example an failures = append(failures, path+"."+key+": example field is absent") continue } - failures = append(failures, comparison.shape(path+"."+key, item, example[key], prepared)...) + failures = append(failures, comparison.shape(path+"."+key, item, example[key], mode)...) } for _, key := range workstreamWireKeys(actual) { if _, present := example[key]; !present { @@ -272,9 +282,12 @@ func (comparison *workstreamWireComparison) shape(path string, value, example an } case []any: actual := value.([]any) - if prepared && len(actual) != len(example) { + if mode == workstreamWirePreparedShape && len(actual) != len(example) { failures = append(failures, fmt.Sprintf("%s: preparation changed array length from %d to %d", path, len(example), len(actual))) } + if mode == workstreamWireReplyShape && len(actual) < len(example) { + failures = append(failures, fmt.Sprintf("%s: want at least %d example elements, got %d", path, len(example), len(actual))) + } if len(example) == 0 { return failures } @@ -290,12 +303,12 @@ func (comparison *workstreamWireComparison) shape(path string, value, example an index = i } itemPath := fmt.Sprintf("%s[%d]", path, i) - differences := comparison.shape(itemPath, item, example[index], prepared) - if _, question := comparison.Questions[itemPath]; !prepared && question && i < len(example) && len(differences) > 0 { + differences := comparison.shape(itemPath, item, example[index], mode) + if _, question := comparison.Questions[itemPath]; mode != workstreamWirePreparedShape && question && i < len(example) && len(differences) > 0 { comparison.Differed[itemPath] = true // The question permits only the shape explicitly illustrated for // this element, so it cannot hide new keys, nulls or wrong kinds. - differences = comparison.shape(itemPath, item, example[i], true) + differences = comparison.shape(itemPath, item, example[i], workstreamWirePreparedShape) } failures = append(failures, differences...) } @@ -355,11 +368,14 @@ type workstreamWireUnavailable struct{ *appBackend } func (*workstreamWireUnavailable) Warm() {} func (*workstreamWireUnavailable) ServesWorkstreams() bool { return false } +const workstreamWireTimeout = 30 * time.Second + type workstreamWireConnection struct { - In *io.PipeWriter - Lines <-chan []byte - Err <-chan error - Next int + In *io.PipeWriter + Lines <-chan []byte + Err <-chan error + Next int + Writes sync.WaitGroup } func workstreamWireConnect(t *testing.T, backend appwire.Engine) *workstreamWireConnection { @@ -370,8 +386,10 @@ func workstreamWireConnect(t *testing.T, backend appwire.Engine) *workstreamWire done := make(chan error, 1) server := appwire.NewServer(outW) go func() { - done <- server.Serve(ctx, inR, appwire.Hello{Version: "test"}, backend) + err := server.Serve(ctx, inR, appwire.Hello{Version: "test"}, backend) + _ = inR.Close() _ = outW.Close() + done <- err }() lines := make(chan []byte, 32) readErr := make(chan error, 1) @@ -390,14 +408,16 @@ func workstreamWireConnect(t *testing.T, backend appwire.Engine) *workstreamWire } readErr <- scanner.Err() }() + connection := &workstreamWireConnection{In: inW, Lines: lines, Err: readErr} t.Cleanup(func() { _ = inW.Close() + connection.Writes.Wait() select { case err := <-done: if err != nil { t.Errorf("wire Serve: %v", err) } - case <-time.After(30 * time.Second): + case <-time.After(workstreamWireTimeout): t.Error("wire Serve did not close the backend") } cancel() @@ -409,7 +429,6 @@ func workstreamWireConnect(t *testing.T, backend appwire.Engine) *workstreamWire t.Error("wire reader did not stop") } }) - connection := &workstreamWireConnection{In: inW, Lines: lines, Err: readErr} hello := connection.read(t) if hello["event"] != appwire.EventHello { t.Fatalf("want hello before replies, got %v", hello) @@ -429,7 +448,7 @@ func (connection *workstreamWireConnection) read(t *testing.T) map[string]any { t.Fatalf("wire frame is not an object: %s", raw) } return frame - case <-time.After(30 * time.Second): + case <-time.After(workstreamWireTimeout): t.Fatal("wire reply timed out") return nil } @@ -439,8 +458,20 @@ func (connection *workstreamWireConnection) call(t *testing.T, method string, pa t.Helper() connection.Next++ raw := workstreamWireEncode(t, map[string]any{"id": connection.Next, "method": method, "params": params}) - if _, err := connection.In.Write(append(raw, '\n')); err != nil { - t.Fatal(err) + writeErr := make(chan error, 1) + connection.Writes.Add(1) + go func() { + defer connection.Writes.Done() + _, err := connection.In.Write(append(raw, '\n')) + writeErr <- err + }() + select { + case err := <-writeErr: + if err != nil { + t.Fatal(err) + } + case <-time.After(workstreamWireTimeout): + t.Fatal("wire request write timed out") } for { frame := connection.read(t) @@ -635,7 +666,7 @@ func (fixture *workstreamWireFixture) grant(t *testing.T) { } // The decoded example is a fresh tree on every call. Only existing values are -// replaced; shape(..., true) independently proves that keys and kinds survive. +// replaced; prepared shape comparison proves that keys and kinds survive. func (fixture *workstreamWireFixture) prepare(t *testing.T, raw json.RawMessage) map[string]any { t.Helper() var replace func(any) any @@ -707,7 +738,7 @@ func (contract *workstreamWireContract) checkError(t *testing.T, path string, va } if example != nil { comparison := &workstreamWireComparison{} - workstreamWireCheck(t, comparison.shape(path, object, example, false)) + workstreamWireCheck(t, comparison.shape(path, object, example, workstreamWireReplyShape)) if exactMessage && object["message"] != example.(map[string]any)["message"] { t.Errorf("%s.message: got %q, want %q", path, message, example.(map[string]any)["message"]) } @@ -811,6 +842,45 @@ func TestAppWorkstreamWireConformance(t *testing.T) { } }) + // (b), (d): missing reply tails cannot skip illustrated shapes; events may + // report fewer changed parts, and prepared arrays must retain exact lengths. + t.Run("array_coverage", func(t *testing.T) { + example := `[{"job":1},{"draft":"draft"}]` + for _, rule := range []struct { + Name string + JSON string + Example string + Mode workstreamWireShapeMode + Valid bool + }{ + {"same_index", example, example, workstreamWireReplyShape, true}, + {"missing_tail", `[{"job":1}]`, example, workstreamWireReplyShape, false}, + {"empty_reply", `[]`, example, workstreamWireReplyShape, false}, + {"nested_missing_tail", `{"items":[{"job":1}]}`, `{"items":` + example + `}`, workstreamWireReplyShape, false}, + {"additional_first_shape", `[{"job":1},{"draft":"draft"},{"job":2}]`, example, workstreamWireReplyShape, true}, + {"additional_wrong_shape", `[{"job":1},{"draft":"draft"},{"draft":"other"}]`, example, workstreamWireReplyShape, false}, + {"empty_example", `[{"job":1}]`, `[]`, workstreamWireReplyShape, true}, + {"prepared_short", `[{"job":1}]`, example, workstreamWirePreparedShape, false}, + {"prepared_long", `[{"job":1},{"draft":"draft"},{"job":2}]`, example, workstreamWirePreparedShape, false}, + {"event_fewer_parts", `{"parts":["pages"]}`, `{"parts":["pages","plan"]}`, workstreamWireEventShape, true}, + {"event_empty_parts", `{"parts":[]}`, `{"parts":["pages","plan"]}`, workstreamWireEventShape, false}, + } { + t.Run(rule.Name, func(t *testing.T) { + path := "array coverage " + rule.Name + " " + comparison := &workstreamWireComparison{} + failures := comparison.shape(path, workstreamWireDecode(t, []byte(rule.JSON)), workstreamWireDecode(t, []byte(rule.Example)), rule.Mode) + if (len(failures) == 0) != rule.Valid { + t.Errorf("valid = %v, want %v: %v", len(failures) == 0, rule.Valid, failures) + } + for _, failure := range failures { + if !strings.HasPrefix(failure, path) { + t.Errorf("failure lost its JSON path: %s", failure) + } + } + }) + } + }) + // (b), (d): malformed examples are contract failures, before any engine call. t.Run("doc_self_check", func(t *testing.T) { for _, name := range cases { @@ -872,7 +942,7 @@ func TestAppWorkstreamWireConformance(t *testing.T) { fixture := workstreamWireSeed(t, name, i) params := fixture.prepare(t, examples.Examples.Params[paramsIndex]) paramsPath := fmt.Sprintf("%s params[%d] ", name, paramsIndex) - workstreamWireCheck(t, comparison.shape(paramsPath, params, workstreamWireDecode(t, examples.Examples.Params[paramsIndex]), true)) + workstreamWireCheck(t, comparison.shape(paramsPath, params, workstreamWireDecode(t, examples.Examples.Params[paramsIndex]), workstreamWirePreparedShape)) if examples.ParamsType != "" { workstreamWireCheck(t, contract.conforms(paramsPath, params, examples.ParamsType)) } @@ -883,7 +953,14 @@ func TestAppWorkstreamWireConformance(t *testing.T) { workstreamWireCheck(t, contract.conforms(resultPath, result, examples.ResultType)) } workstreamWireCheck(t, workstreamWireNoNulls(resultPath, result)) - workstreamWireCheck(t, comparison.shape(resultPath, result, workstreamWireDecode(t, examples.Examples.Result[resultIndex]), false)) + workstreamWireCheck(t, comparison.shape(resultPath, result, workstreamWireDecode(t, examples.Examples.Result[resultIndex]), workstreamWireReplyShape)) + if name == "page.delete" { + deleted, ok := result.(map[string]any)["deleted"].([]any) + want := []any{fixture.IDs["pg_32e1"], fixture.IDs["conformance-descendant"]} + if !ok || len(deleted) != len(want) || !slices.Contains(deleted, want[0]) || !slices.Contains(deleted, want[1]) { + t.Errorf("%s.deleted: want exactly seeded parent and descendant %v, got %v", resultPath, want, deleted) + } + } if name == "collections.add" { workstreamWireCollectionResult(t, result) } @@ -1066,7 +1143,7 @@ func workstreamWireEvent(t *testing.T, contract *workstreamWireContract, example if exampleIndex >= len(examples.Examples.Data) { t.Fatalf("%s: no example for parts presence", path) } - workstreamWireCheck(t, comparison.shape(path, value, workstreamWireDecode(t, examples.Examples.Data[exampleIndex]), false)) + workstreamWireCheck(t, comparison.shape(path, value, workstreamWireDecode(t, examples.Examples.Data[exampleIndex]), workstreamWireEventShape)) if event.ID != fixture.Stream.ID { t.Errorf("%s.id: got %s, want seeded workstream %s", path, event.ID, fixture.Stream.ID) } @@ -1117,7 +1194,7 @@ func workstreamWireError(t *testing.T, contract *workstreamWireContract, files m } if errorCase.Name != "bad_params" && errorCase.Name != "too_large" { comparison := &workstreamWireComparison{} - workstreamWireCheck(t, comparison.shape(errorCase.Method+" error params ", params, workstreamWireDecode(t, examples.Examples.Params[0]), true)) + workstreamWireCheck(t, comparison.shape(errorCase.Method+" error params ", params, workstreamWireDecode(t, examples.Examples.Params[0]), workstreamWirePreparedShape)) } connection := workstreamWireConnect(t, engine) frame := connection.call(t, errorCase.Method, params)