diff --git a/.github/workflows/python-integration.yml b/.github/workflows/python-integration.yml index d45885a..2c380e3 100644 --- a/.github/workflows/python-integration.yml +++ b/.github/workflows/python-integration.yml @@ -4,10 +4,12 @@ on: push: paths: - "python/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/python-integration.yml" pull_request: paths: - "python/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/python-integration.yml" workflow_dispatch: inputs: @@ -24,6 +26,9 @@ jobs: - name: Checkout cccc-sdk uses: actions/checkout@v4 + - name: Check SDK hardening contract + run: python3 scripts/check_sdk_hardening.py + - name: Checkout cccc (daemon) uses: actions/checkout@v4 with: @@ -54,7 +59,7 @@ jobs: python -m pip install -U pip python -m pip install -e python - - name: Smoke (daemon + sdk) + - name: Smoke current message and Mail contract run: | set -euo pipefail export CCCC_HOME="$PWD/.tmp_cccc_home" @@ -65,3 +70,46 @@ jobs: trap cleanup EXIT cccc daemon start python python/examples/compat_check.py + + python - <<'PY' + from cccc_sdk import CCCCClient + + client = CCCCClient() + created = client.group_create(title="python-sdk-native-current") + group_id = str(created.get("group_id") or "") + if not group_id: + raise RuntimeError("group_create did not return group_id") + + try: + client.group_start(group_id=group_id) + client.actor_add( + group_id=group_id, + actor_id="sdk-live", + runtime="custom", + runner="headless", + command=["/usr/bin/true"], + ) + options = { + "group_id": group_id, + "text": "python SDK Mail replay probe", + "message_mode": "mail", + "to": ["sdk-live"], + "client_id": f"python-sdk:{group_id}:mail", + } + first = client.send(**options) + replay = client.send(**options) + if (first.get("event") or {}).get("id") != (replay.get("event") or {}).get("id"): + raise RuntimeError("stable client_id created two Mail events") + if replay.get("duplicate") is not True: + raise RuntimeError("Mail replay did not expose duplicate=true") + + peeked = client.inbox_peek(group_id=group_id, actor_id="sdk-live", by="sdk-live") + consumed = client.inbox_read(group_id=group_id, actor_id="sdk-live", by="sdk-live") + if len(peeked.get("messages") or []) != 1 or len(consumed.get("messages") or []) != 1: + raise RuntimeError("Mail was not visible and consumed exactly once") + empty = client.inbox_read(group_id=group_id, actor_id="sdk-live", by="sdk-live") + if empty.get("messages"): + raise RuntimeError("Mail remained unread after inbox_read") + finally: + client.group_delete(group_id=group_id) + PY diff --git a/.github/workflows/rust-ci.yml b/.github/workflows/rust-ci.yml index 3bb0f72..fe9ee85 100644 --- a/.github/workflows/rust-ci.yml +++ b/.github/workflows/rust-ci.yml @@ -4,10 +4,12 @@ on: push: paths: - "rust/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/rust-ci.yml" pull_request: paths: - "rust/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/rust-ci.yml" jobs: @@ -15,6 +17,8 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - name: Check SDK hardening contract + run: python3 scripts/check_sdk_hardening.py - uses: dtolnay/rust-toolchain@stable with: components: clippy,rustfmt @@ -23,17 +27,39 @@ jobs: run: cargo fmt --check - name: Lint working-directory: rust - run: cargo clippy --all-targets --all-features -- -D warnings + run: cargo clippy --locked --all-targets --all-features -- -D warnings - name: Test working-directory: rust - run: cargo test --all-targets + run: cargo test --locked --all-targets - name: Package working-directory: rust run: cargo package --locked + msrv: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@master + with: + toolchain: "1.74.0" + - name: Check declared MSRV + working-directory: rust + run: cargo check --locked + + windows: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - name: Test Windows client behavior + working-directory: rust + run: cargo test --locked --all-targets + integration: runs-on: ubuntu-latest needs: test + env: + CCCC_RUN_LIVE_RELIABILITY: "1" steps: - uses: actions/checkout@v4 @@ -73,4 +99,5 @@ jobs: trap cleanup EXIT cccc daemon start test -f "$CCCC_HOME/daemon/ccccd.addr.json" - cargo run --manifest-path rust/Cargo.toml --example compat_check + cargo run --locked --manifest-path rust/Cargo.toml --example compat_check + cargo test --locked --manifest-path rust/Cargo.toml --test live_reliability -- --nocapture diff --git a/.github/workflows/spec-drift.yml b/.github/workflows/spec-drift.yml index 7e3f1f5..3e7ac96 100644 --- a/.github/workflows/spec-drift.yml +++ b/.github/workflows/spec-drift.yml @@ -5,11 +5,13 @@ on: paths: - "spec/**" - "scripts/check_specs_against_cccc.sh" + - "scripts/check_sdk_hardening.py" - ".github/workflows/spec-drift.yml" pull_request: paths: - "spec/**" - "scripts/check_specs_against_cccc.sh" + - "scripts/check_sdk_hardening.py" - ".github/workflows/spec-drift.yml" schedule: - cron: "17 3 * * *" @@ -36,3 +38,6 @@ jobs: - name: Compare mirrored standards run: bash scripts/check_specs_against_cccc.sh cccc-core + + - name: Check SDK hardening contract + run: python3 scripts/check_sdk_hardening.py diff --git a/.github/workflows/ts-ci.yml b/.github/workflows/ts-ci.yml index fc3f4d7..4bf358c 100644 --- a/.github/workflows/ts-ci.yml +++ b/.github/workflows/ts-ci.yml @@ -4,10 +4,12 @@ on: push: paths: - "ts/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/ts-ci.yml" pull_request: paths: - "ts/**" + - "scripts/check_sdk_hardening.py" - ".github/workflows/ts-ci.yml" jobs: @@ -17,6 +19,9 @@ jobs: steps: - uses: actions/checkout@v4 + - name: Check SDK hardening contract + run: python3 scripts/check_sdk_hardening.py + - uses: actions/setup-node@v4 with: node-version: "20" @@ -75,7 +80,7 @@ jobs: working-directory: cccc run: cargo build --locked -p cccc --bin cccc - - name: SDK <-> daemon integration smoke + - name: SDK against current message and Mail contract run: | set -euo pipefail export CCCC_HOME="$PWD/.tmp_cccc_home" @@ -95,14 +100,13 @@ jobs: import fs from 'node:fs/promises'; import { CCCCClient, discoverEndpoint } from './ts/dist/index.js'; - // Validate daemon TCP endpoint normalization for 0.0.0.0/localhost style descriptors. const fakeHome = await fs.mkdtemp(path.join(os.tmpdir(), 'cccc-sdk-ts-')); const fakeDaemon = path.join(fakeHome, 'daemon'); await fs.mkdir(fakeDaemon, { recursive: true }); await fs.writeFile( path.join(fakeDaemon, 'ccccd.addr.json'), JSON.stringify({ v: 1, transport: 'tcp', host: '0.0.0.0', port: 12345 }), - 'utf-8' + 'utf-8', ); const fakeEndpoint = await discoverEndpoint(fakeHome); if (fakeEndpoint.host !== '127.0.0.1') { @@ -120,7 +124,11 @@ jobs: 'group_preamble_set', 'group_preamble_reset', 'send', + 'reply', 'send_files', + 'inbox_peek', + 'inbox_read', + 'message_history', 'events_stream', 'memory_search', 'memory_get', @@ -140,27 +148,54 @@ jobs: const created = await client.groupCreate({ title: 'ts-integration' }); const groupId = String(created.group_id || ''); - if (!groupId) { - throw new Error('groupCreate did not return group_id'); - } - - const eventPromise = (async () => { - for await (const item of client.eventsStream({ groupId, by: 'user' })) { - if (item.t === 'event') return item.event; + if (!groupId) throw new Error('groupCreate did not return group_id'); + + try { + await client.groupStart(groupId); + await client.actorAdd({ + groupId, + actorId: 'sdk-live', + runtime: 'custom', + runner: 'headless', + command: ['/usr/bin/true'], + }); + + const eventPromise = (async () => { + for await (const item of client.eventsStream({ groupId, by: 'user' })) { + if (item.t === 'event') return item.event; + } + return null; + })(); + await new Promise((resolve) => setTimeout(resolve, 200)); + await client.send({ groupId, text: 'hello from ts-ci', mode: 'send', to: ['user'] }); + const event = await Promise.race([ + eventPromise, + new Promise((_, reject) => setTimeout(() => reject(new Error('stream timeout')), 5000)), + ]); + if (!event || typeof event.id !== 'string' || event.id.length === 0) { + throw new Error('stream event is missing canonical id'); } - return null; - })(); - - await new Promise((r) => setTimeout(r, 200)); - await client.send({ groupId, text: 'hello from ts-ci', mode: 'send', by: 'user', to: ['user'] }); - const timeoutPromise = new Promise((_, reject) => - setTimeout(() => reject(new Error('timed out waiting for stream event')), 5000) - ); - const event = await Promise.race([eventPromise, timeoutPromise]); - - if (!event || typeof event.id !== 'string' || event.id.length === 0) { - throw new Error('stream event is missing canonical id field'); + const mailOptions = { + groupId, + text: 'TS SDK Mail replay probe', + mode: 'mail', + to: ['sdk-live'], + clientId: `ts-sdk:${groupId}:mail`, + }; + const first = await client.send(mailOptions); + const replay = await client.send(mailOptions); + if (first.event?.id !== replay.event?.id || replay.duplicate !== true) { + throw new Error('stable clientId did not replay the Mail event'); + } + const peeked = await client.inboxPeek({ groupId, actorId: 'sdk-live', by: 'sdk-live' }); + const consumed = await client.inboxRead({ groupId, actorId: 'sdk-live', by: 'sdk-live' }); + if (peeked.messages?.length !== 1 || consumed.messages?.length !== 1) { + throw new Error('Mail was not visible and consumed exactly once'); + } + const empty = await client.inboxRead({ groupId, actorId: 'sdk-live', by: 'sdk-live' }); + if (empty.messages?.length) throw new Error('Mail remained unread after inboxRead'); + } finally { + await client.groupDelete(groupId); } - console.log('integration smoke passed', { groupId, eventId: event.id }); JS diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cae9e7..359ea6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ CCCC line and exposes the IPC surface available on that line. - Reduced legacy Group Space synchronization to its contractually supported read-only status operation. Explicit ingest and source operations remain the mutation path. +- Reworked the Rust identity-bound adapter around explicit message modes, + stable `client_id` replay, and the daemon's atomic Mail Inbox transaction; + removed its emulation of retired ACK/read cursor operations. ### Compatibility @@ -28,8 +31,9 @@ CCCC line and exposes the IPC surface available on that line. implementation label alone. - Internal Web upload preflight and relay-only operations remain available via generic calls but intentionally have no first-class public wrapper. -- Package versions are deliberately unchanged. Version selection and publishing - remain a separate release decision. +- Python and TypeScript package versions remain unchanged. The Rust hardening + branch keeps its already-selected `0.0.2` source version; publishing remains a + separate release decision. ## [0.4.34] — Unreleased @@ -45,6 +49,10 @@ CCCC line and exposes the IPC surface available on that line. after an exchange has begun. - Scheduled CI drift detection for all three mirrored CCCC standards, automatic Python integration coverage, and a live Rust-SDK/current-daemon smoke job. +- Python and TypeScript Web Model wait/complete helpers with a required stable + `delivery_id` replay key. +- Rust 0.0.2 identity-bound reliable messaging with explicit Send / Send + Reply + / Mail modes, stable write keys, and atomic Mail consumption. ### Changed @@ -80,6 +88,10 @@ CCCC line and exposes the IPC surface available on that line. - TypeScript's `INVALID_REQUEST` constant now matches the daemon's `invalid_request` code and includes current request-size and Remote Access administrator-token errors. +- TypeScript connection, response, and stream-handshake phases now share one + deadline; handshake cancellation closes the socket, buffered stream data is + byte-capped before decoding, and transport cleanup removes only SDK-owned + listeners. - The daemon IPC mirror now matches current CCCC core, including Remote Access administrator-token state and enforcement fields. @@ -91,6 +103,17 @@ CCCC line and exposes the IPC surface available on that line. maps and side-effect-free compatibility probing. TypeScript also compiles exported option fixtures so documented contract values cannot silently drift out of the published declaration surface. +- Added current native-daemon integration for message replay and atomic Mail + consumption, Rust 1.74 MSRV and Windows jobs, and a static SDK hardening gate. + +## Rust crate [0.0.2] — Unreleased + +### Added + +- Least-privilege `IdentityBoundClient` adapters for explicit-mode idempotent + send/reply and daemon-owned Mail Inbox operations. +- Nullable native cursor decoding and direct use of atomic `inbox_read`, without + retaining the retired all-message mark-read compatibility model. ## Rust crate [0.0.1] — 2026-08-03 diff --git a/README.ja.md b/README.ja.md index a43f1f2..13e053b 100644 --- a/README.ja.md +++ b/README.ja.md @@ -29,6 +29,7 @@ SDK と CCCC Web が同じ `CCCC_HOME` を参照していれば、書き込み 主な用途： - リアルタイム更新が必要な Web/IDE プラグイン（`events_stream`） - Working Group を監視して自動応答する bot/service +- identity-bound な冪等 write と atomic Mail 消費を必要とする高信頼 Rust worker - group / actors / shared context / capability ポリシー / Group Space をプログラムから管理する社内ツール - `tracked_send`、Context Ops v3 task/agent state、capability discovery、ローカル memory API を使う workflow 連携 @@ -90,7 +91,7 @@ python python/examples/send.py --group g_xxx --text "FYI" --mode mail ```toml [dependencies] -cccc-sdk = "0.0.1" +cccc-sdk = "0.0.2" ``` Rust クライアントは `CCCC_HOME` の Unix Socket/TCP daemon を自動検出し、 @@ -102,7 +103,7 @@ Rust クライアントは `CCCC_HOME` の Unix Socket/TCP daemon を自動検 ## バージョニングと互換性 SDK リリースは daemon のバージョン文字列ではなく contract に追従します： -- Python と TypeScript は現在の SDK リリースラインに追従し、Rust crate は `0.0.1` から開始します。 +- Python と TypeScript は現在の SDK リリースラインに追従し、Rust crate は当面 `0.0.x` 系列です。 - 実行時互換性は `assert_compatible(...)` で必要な capability/op を指定して確認します。 互換性は “契約/能力” で保証し、バージョン文字列の厳密一致には依存しません： diff --git a/README.md b/README.md index ea66d71..d348b51 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ If SDK clients and CCCC Web use the same `CCCC_HOME`, all writes are shared imme Typical use cases: - Reactive UI / IDE plugins that need real-time updates (`events_stream`) - Bots/services that watch groups and respond automatically +- Reliable Rust workers that need identity-bound idempotent writes and atomic Mail consumption - Internal tools that create/manage groups, actors, shared context, capability policy, and Group Space programmatically - Workflow integrations that use `tracked_send`, Context Ops v3 task/agent state updates, capability discovery, and first-class local memory @@ -93,7 +94,7 @@ python python/examples/send.py --group g_xxx --text "FYI" --mode mail ```toml [dependencies] -cccc-sdk = "0.0.1" +cccc-sdk = "0.0.2" ``` ```rust @@ -117,7 +118,7 @@ fn main() -> Result<(), Box> { SDK releases follow daemon contracts, not strict daemon version strings: - Python and TypeScript package versions track the current SDK release line; the - Rust crate starts at `0.0.1` while its public API settles. + Rust crate is on the `0.0.x` line while its public API settles. - Use `assert_compatible(...)` with required capabilities/ops for runtime gating. Compatibility is enforced by **contracts**, not by strict version string matching: @@ -138,6 +139,8 @@ This repo keeps a mirror under `spec/`: ```bash ./scripts/sync_specs_from_cccc.sh ../cccc +# Reproduce a tagged mirror when auditing an older contract: +./scripts/sync_specs_from_cccc.sh ../cccc v0.4.35 ``` The sync command intentionally replaces only the three mirrored standards. diff --git a/README.zh-CN.md b/README.zh-CN.md index 5d2083e..a293c4c 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -29,6 +29,7 @@ CCCC SDK 是一套用于 CCCC 平台的**客户端 SDK**。 典型场景： - 需要实时更新的 Web/IDE 插件（`events_stream`） - 监听工作组并自动响应的 bot/service +- 需要身份绑定幂等写入与原子 Mail 消费的可靠 Rust worker - 以编程方式创建/管理 group、actors、共享 context、capability 策略与 Group Space 的内部工具 - 使用 `tracked_send`、Context Ops v3 任务/agent state、capability discovery、本地 memory API 的工作流集成 @@ -90,7 +91,7 @@ python python/examples/send.py --group g_xxx --text "FYI" --mode mail ```toml [dependencies] -cccc-sdk = "0.0.1" +cccc-sdk = "0.0.2" ``` Rust 客户端会自动发现 `CCCC_HOME` 下的 Unix Socket/TCP daemon，并提供通用 @@ -101,7 +102,7 @@ Rust 客户端会自动发现 `CCCC_HOME` 下的 Unix Socket/TCP daemon，并提 ## 版本策略与兼容性 SDK 发布跟随 daemon 合约，而不是硬匹配 daemon 版本号： -- Python 和 TypeScript 包版本跟随当前 SDK 发布线；Rust crate 从 `0.0.1` 起步。 +- Python 和 TypeScript 包版本跟随当前 SDK 发布线；Rust crate 暂处于 `0.0.x` 版本线。 - 运行时兼容请用 `assert_compatible(...)` 指定所需 capability/op。 我们保证兼容性的手段是“契约/能力”，而不是字符串版本号硬匹配： diff --git a/RELEASING.md b/RELEASING.md index 60b5d7f..436c145 100644 --- a/RELEASING.md +++ b/RELEASING.md @@ -11,7 +11,8 @@ This repo is a monorepo with three deliverables: not choose or modify the next package version. - RC sequence is SDK-owned (PEP 440 `X.Y.ZrcN` for Python and SemVer `X.Y.Z-rc.N` for npm). -- The Rust crate begins at `0.0.1` while its public API settles. +- The Rust crate begins at `0.0.1`; the current hardening branch prepares + `0.0.2`, but publishing remains a separate release decision. - Compatibility is enforced by contracts/capabilities/op-probing, not by matching RC numbers. ## 0) Sync specs (recommended) @@ -19,6 +20,9 @@ This repo is a monorepo with three deliverables: ```bash ./scripts/sync_specs_from_cccc.sh ../cccc ./scripts/check_specs_against_cccc.sh ../cccc +# Reproducible audit against a committed core revision: +./scripts/check_specs_against_cccc.sh ../cccc +python3 scripts/check_sdk_hardening.py ``` ## 1) Python release (PyPI/TestPyPI) @@ -103,11 +107,15 @@ npm publish --access public ```bash cd rust cargo fmt --check -cargo clippy --all-targets --all-features -- -D warnings -cargo test --all-targets +cargo clippy --locked --all-targets --all-features -- -D warnings +cargo test --locked --all-targets +cargo +1.74.0 check --locked cargo package --locked ``` +The Rust CI matrix also runs the full suite on Windows and the opt-in reliable +messaging test against the current native CCCC daemon. + ### Publish ```bash diff --git a/python/README.md b/python/README.md index 8936c5c..6005eaf 100644 --- a/python/README.md +++ b/python/README.md @@ -249,8 +249,28 @@ recovered = c.web_model_runtime_recover_turn( actor_id="web-model", event_ids=["e_xxx"], ) + +# Complete a runtime-owned turn. Reuse the exact delivery_id if the caller +# must retry after an unknown outcome. +turn = c.web_model_runtime_wait_next_turn( + group_id="g_xxx", + actor_id="web-model", +) +payload = turn["turn"] +delivery_id = f"worker:{payload['turn_id']}" +c.web_model_runtime_complete_turn( + group_id="g_xxx", + actor_id="web-model", + turn_id=payload["turn_id"], + delivery_id=delivery_id, + event_ids=payload["event_ids"], +) ``` +`delivery_id` is required and is the completion replay key. A retry for the +same acquired turn must use the same value; generating a new value can create a +second completion receipt. + `term_resize()` sends the standard `term_resize` operation. For older compatible daemon builds that expose `terminal_resize`, the SDK falls back only after receiving a structured `unknown_op`; transport failures are never diff --git a/python/src/cccc_sdk/client_0430_ops.py b/python/src/cccc_sdk/client_0430_ops.py index 7b69ae1..1cab2a7 100644 --- a/python/src/cccc_sdk/client_0430_ops.py +++ b/python/src/cccc_sdk/client_0430_ops.py @@ -3,11 +3,13 @@ from .client_0430_admin_ops import CCCC0430AdminOpsMixin from .client_0430_assistant_ops import CCCC0430AssistantOpsMixin from .client_0430_memory_ops import CCCC0430MemoryOpsMixin +from .client_0430_runtime_ops import CCCC0430RuntimeOpsMixin class CCCC0430OpsMixin( CCCC0430AdminOpsMixin, CCCC0430AssistantOpsMixin, CCCC0430MemoryOpsMixin, + CCCC0430RuntimeOpsMixin, ): pass diff --git a/python/src/cccc_sdk/client_0430_runtime_ops.py b/python/src/cccc_sdk/client_0430_runtime_ops.py new file mode 100644 index 0000000..6eb2fa5 --- /dev/null +++ b/python/src/cccc_sdk/client_0430_runtime_ops.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .client_0430_shared import _compact + + +class CCCC0430RuntimeOpsMixin: + """Web Model runtime operations supported by the current native contract.""" + + def web_model_runtime_wait_next_turn( + self, + *, + group_id: str, + actor_id: str, + by: Optional[str] = None, + limit: int = 20, + kind_filter: str = "all", + ) -> Dict[str, Any]: + aid = str(actor_id) + return self.call( + "web_model_runtime_wait_next_turn", + { + "group_id": str(group_id), + "actor_id": aid, + "by": str(by) if by is not None else aid, + "limit": min(max(int(limit), 1), 20), + "kind_filter": str(kind_filter), + }, + ) + + def web_model_runtime_complete_turn( + self, + *, + group_id: str, + actor_id: str, + turn_id: str, + delivery_id: str, + event_ids: Optional[List[str]] = None, + latest_event_id: str = "", + status: str = "done", + summary: str = "", + by: Optional[str] = None, + ) -> Dict[str, Any]: + aid = str(actor_id) + return self.call( + "web_model_runtime_complete_turn", + _compact( + { + "group_id": str(group_id), + "actor_id": aid, + "by": str(by) if by is not None else aid, + "turn_id": str(turn_id), + "delivery_id": str(delivery_id), + "event_ids": [str(event_id) for event_id in event_ids] + if event_ids is not None + else None, + "latest_event_id": latest_event_id or None, + "status": str(status), + "summary": summary or None, + } + ), + ) diff --git a/python/tests/test_client_0430_contract.py b/python/tests/test_client_0430_contract.py index 616e5b6..cb8351f 100644 --- a/python/tests/test_client_0430_contract.py +++ b/python/tests/test_client_0430_contract.py @@ -8,7 +8,7 @@ from cccc_sdk.transport import DaemonEndpoint -class TestClient0433Contract(unittest.TestCase): +class TestCurrentNativeContract(unittest.TestCase): def _client(self) -> CCCCClient: return CCCCClient(endpoint=DaemonEndpoint(transport="tcp", host="127.0.0.1", port=9000)) @@ -350,6 +350,43 @@ def fake_call_daemon(*, endpoint, request, timeout_s): # type: ignore[no-untype self.assertEqual(captured[1]["args"]["text"], "Include omissions") self.assertEqual(captured[1]["args"]["source_text"], "Current meeting notes") + def test_web_model_completion_requires_and_reuses_delivery_id(self) -> None: + captured: list[dict] = [] + args = { + "group_id": "g_1", + "actor_id": "web-model", + "turn_id": "turn-1", + "delivery_id": "worker:turn-1", + "event_ids": ["e_1"], + "status": "done", + } + + def fake_call_daemon(*, endpoint, request, timeout_s): # type: ignore[no-untyped-def] + captured.append(request) + return { + "ok": True, + "result": {"delivery_id": args["delivery_id"], "duplicate": True}, + } + + with patch("cccc_sdk.client.call_daemon", side_effect=fake_call_daemon): + client = self._client() + for _ in range(2): + client.web_model_runtime_complete_turn( + group_id=args["group_id"], + actor_id=args["actor_id"], + turn_id=args["turn_id"], + delivery_id=args["delivery_id"], + event_ids=args["event_ids"], + status=args["status"], + ) + + self.assertEqual(captured[0], captured[1]) + self.assertEqual(captured[0]["op"], "web_model_runtime_complete_turn") + self.assertTrue( + {"group_id", "actor_id", "turn_id", "delivery_id"}.issubset(captured[0]["args"]) + ) + self.assertEqual(captured[0]["args"]["delivery_id"], args["delivery_id"]) + if __name__ == "__main__": unittest.main() diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 73bf675..e6ed332 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -4,7 +4,7 @@ version = 3 [[package]] name = "cccc-sdk" -version = "0.0.1" +version = "0.0.2" dependencies = [ "serde", "serde_json", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 3fbcbd6..726eeed 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cccc-sdk" -version = "0.0.1" +version = "0.0.2" edition = "2021" rust-version = "1.74" description = "Official Rust client SDK for CCCC daemon IPC v1" diff --git a/rust/README.md b/rust/README.md index c82017a..e7b83d6 100644 --- a/rust/README.md +++ b/rust/README.md @@ -8,7 +8,7 @@ the same daemon contract. ```toml [dependencies] -cccc-sdk = "0.0.1" +cccc-sdk = "0.0.2" ``` ## Quick start @@ -87,10 +87,29 @@ a connection-establishment failure. Once request exchange begins, failures are reported as `Error::OutcomeUnknown` and are never replayed automatically. Clients created with `new(endpoint)` keep that explicit endpoint. +## Identity-bound reliable messaging + +Rust 0.0.2 adds an `IdentityBoundClient` that exposes only current idempotent +message writes and Mail Inbox operations. A caller supplies a +`WorkloadIdentityHook`; the adapter binds the principal and evidence to every +request instead of treating the wire-level `by` field as authentication. The +receiving daemon or gateway remains responsible for verifying that evidence. + +`send_idempotent` requires an explicit `MessageMode`, recipients, and stable +`client_id`; `reply_idempotent` similarly requires a Send-or-Mail reply mode. +After `Error::OutcomeUnknown`, retry with the exact same arguments and key so +the daemon can return the original event with `replayed=true`. + +Mail consumption uses the daemon's atomic `inbox_read` transaction. The SDK +does not recreate the retired `inbox_list`, `inbox_mark_read`, or generic ACK +model, and it does not maintain a second competing cursor. `inbox_peek` remains +available for non-consuming inspection. The adapter intentionally has no +generic `call`, shutdown, configuration, or credential methods. + `assert_compatible` probes requested operation names and rejects an advertised capability whose actual operation returns `unknown_op`. Streaming upgrade operations such as `events_stream` and `term_attach` are not -exposed as iterators in 0.0.1. `assert_compatible` deliberately skips unsafe +exposed as iterators in 0.0.2. `assert_compatible` deliberately skips unsafe duplex probes; a reusable stream API will be added only with stable ownership, close, and backpressure semantics. diff --git a/rust/src/error.rs b/rust/src/error.rs index 5ebd812..ec41d13 100644 --- a/rust/src/error.rs +++ b/rust/src/error.rs @@ -60,6 +60,9 @@ pub enum Error { #[error("incompatible CCCC daemon: {0}")] Incompatible(String), + + #[error("write reconciliation requires operator action: {0}")] + ReconciliationRequired(String), } pub type Result = std::result::Result; diff --git a/rust/src/identity.rs b/rust/src/identity.rs new file mode 100644 index 0000000..ab5a0f1 --- /dev/null +++ b/rust/src/identity.rs @@ -0,0 +1,99 @@ +use serde_json::{Map, Value}; + +use crate::{Error, Result}; + +/// Principal established by a workload identity provider. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AuthenticatedPrincipal { + pub subject: String, + pub issuer: String, + pub evidence_id: Option, +} + +/// Verifiable carrier produced for one request by an identity hook. +#[derive(Clone, Debug, PartialEq)] +pub struct WorkloadIdentityEvidence { + pub carrier_key: String, + pub carrier: Value, +} + +impl WorkloadIdentityEvidence { + pub fn new(carrier_key: impl Into, carrier: Value) -> Result { + let evidence = Self { + carrier_key: carrier_key.into(), + carrier, + }; + if evidence.carrier_key.trim().is_empty() + || evidence.carrier_key == "by" + || evidence.carrier.is_null() + { + return Err(Error::Incompatible( + "workload identity evidence needs a non-by carrier key and non-null value".into(), + )); + } + Ok(evidence) + } +} + +impl AuthenticatedPrincipal { + pub fn new(subject: impl Into, issuer: impl Into) -> Result { + let principal = Self { + subject: subject.into(), + issuer: issuer.into(), + evidence_id: None, + }; + principal.validate()?; + Ok(principal) + } + + fn validate(&self) -> Result<()> { + if self.subject.trim().is_empty() || self.issuer.trim().is_empty() { + return Err(Error::Incompatible( + "authenticated principal subject and issuer must be non-empty".into(), + )); + } + Ok(()) + } +} + +/// Hook for an external workload identity implementation. +/// +/// `evidence` returns a signature, token, nonce, or other carrier for `args`. +/// The receiving daemon or gateway must verify that carrier; the SDK never +/// treats a caller-provided `by` value as authentication. +pub trait WorkloadIdentityHook { + fn principal(&self) -> Result; + + /// Sign or otherwise bind the operation and canonical args to evidence. + fn evidence( + &self, + operation: &str, + args: &Map, + ) -> Result; +} + +pub(crate) fn bind_identity( + hook: &H, + operation: &str, + args: &mut Map, +) -> Result { + let principal = hook.principal()?; + principal.validate()?; + if let Some(claimed) = args.get("by") { + if claimed.as_str() != Some(&principal.subject) { + return Err(Error::Incompatible( + "request by does not match authenticated principal".into(), + )); + } + } + args.insert("by".into(), Value::String(principal.subject.clone())); + let evidence = hook.evidence(operation, args)?; + if args.contains_key(&evidence.carrier_key) { + return Err(Error::Incompatible(format!( + "workload identity carrier would overwrite request field {}", + evidence.carrier_key + ))); + } + args.insert(evidence.carrier_key, evidence.carrier); + Ok(principal) +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 896b31a..d480c6d 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -7,11 +7,14 @@ mod client; mod endpoint; mod error; +mod identity; mod protocol; +mod reliable; pub use client::{CCCCClient, CompatibilityRequirements}; pub use endpoint::{discover_endpoint, DaemonEndpoint}; pub use error::{DaemonError, Error, Result}; +pub use identity::{AuthenticatedPrincipal, WorkloadIdentityEvidence, WorkloadIdentityHook}; pub use protocol::{ ContextDetail, DaemonRequest, DaemonResponse, MessageHistoryMode, MessageMode, PingResult, ReplyMessageMode, TerminalHistoryOptions, TerminalHistoryResult, TerminalResizeResult, @@ -20,3 +23,4 @@ pub use protocol::{ WebModelDeliveryPreferencesResult, WebModelRecoveredTurn, WebModelRecoveredTurnDelivery, WebModelRuntimeRecoverTurnResult, }; +pub use reliable::{Event, IdentityBoundClient, InboxCursor, InboxPage, MessageWriteResult}; diff --git a/rust/src/reliable.rs b/rust/src/reliable.rs new file mode 100644 index 0000000..9785b49 --- /dev/null +++ b/rust/src/reliable.rs @@ -0,0 +1,455 @@ +use serde::{Deserialize, Deserializer, Serialize}; +use serde_json::{Map, Value}; + +use crate::identity::{bind_identity, AuthenticatedPrincipal, WorkloadIdentityHook}; +use crate::{CCCCClient, Error, MessageMode, ReplyMessageMode, Result}; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Event { + #[serde(alias = "event_id")] + pub id: String, + pub ts: String, + pub kind: String, + #[serde(default)] + pub group_id: String, + #[serde(default)] + pub by: String, + #[serde(default)] + pub data: Value, + #[serde(flatten)] + pub extra: Map, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct InboxCursor { + #[serde(default, deserialize_with = "deserialize_nullable_string")] + pub event_id: String, + #[serde(default, deserialize_with = "deserialize_nullable_string")] + pub ts: String, + #[serde(default)] + pub updated_at: Option, +} + +#[derive(Clone, Debug, Deserialize)] +pub struct InboxPage { + #[serde(default)] + pub messages: Vec, + pub cursor: InboxCursor, + #[serde(default)] + pub event: Option, +} + +#[derive(Clone, Debug, Deserialize)] +pub struct MessageWriteResult { + pub event: Event, + pub message_mode: String, + #[serde(default, alias = "duplicate")] + pub replayed: bool, +} + +/// Identity-bound adapter for the current message contract. +/// +/// The adapter deliberately exposes no generic call, daemon lifecycle, +/// configuration, or credential operations. Stable `client_id` values are the +/// reconciliation boundary for writes. Mail reads delegate to the daemon's +/// atomic `inbox_read` transaction instead of emulating the retired ACK/cursor +/// operations. +pub struct IdentityBoundClient { + client: CCCCClient, + identity: H, + principal: AuthenticatedPrincipal, +} + +impl IdentityBoundClient { + pub fn new(client: CCCCClient, identity: H) -> Result { + let principal = identity.principal()?; + if principal.subject.trim().is_empty() || principal.issuer.trim().is_empty() { + return Err(Error::Incompatible( + "authenticated principal subject and issuer must be non-empty".into(), + )); + } + Ok(Self { + client, + identity, + principal, + }) + } + + pub fn principal(&self) -> &AuthenticatedPrincipal { + &self.principal + } + + fn call(&self, operation: &str, mut args: Map) -> Result> { + let current = bind_identity(&self.identity, operation, &mut args)?; + if current != self.principal { + return Err(Error::Incompatible( + "workload identity principal changed during the session".into(), + )); + } + self.client.call(operation, args) + } + + fn call_typed Deserialize<'de>>( + &self, + operation: &str, + args: Map, + ) -> Result { + Ok(serde_json::from_value(Value::Object( + self.call(operation, args)?, + ))?) + } + + /// Append a message with a daemon-stable `client_id`. + /// + /// Reuse the exact mode, recipients, content, and key after + /// [`Error::OutcomeUnknown`]. Never generate a new key for that retry. + pub fn send_idempotent( + &self, + group_id: &str, + text: &str, + message_mode: MessageMode, + recipients: &[&str], + client_id: &str, + ) -> Result { + validate_client_id(client_id)?; + validate_recipients(recipients)?; + let mut args = object([ + ("group_id", Value::String(group_id.into())), + ("text", Value::String(text.into())), + ("message_mode", Value::String(message_mode.as_str().into())), + ("client_id", Value::String(client_id.into())), + ]); + insert_recipients(&mut args, recipients); + self.call_typed("send", args) + } + + pub fn reconcile_send( + &self, + group_id: &str, + text: &str, + message_mode: MessageMode, + recipients: &[&str], + client_id: &str, + ) -> Result { + self.send_idempotent(group_id, text, message_mode, recipients, client_id) + } + + /// Append a reply with a daemon-stable `client_id`. + /// + /// Replies may use Send or Mail but cannot create another reply request. + pub fn reply_idempotent( + &self, + group_id: &str, + reply_to: &str, + text: &str, + message_mode: ReplyMessageMode, + recipients: &[&str], + client_id: &str, + ) -> Result { + validate_client_id(client_id)?; + validate_recipients(recipients)?; + let mut args = object([ + ("group_id", Value::String(group_id.into())), + ("reply_to", Value::String(reply_to.into())), + ("text", Value::String(text.into())), + ("message_mode", Value::String(message_mode.as_str().into())), + ("client_id", Value::String(client_id.into())), + ]); + insert_recipients(&mut args, recipients); + self.call_typed("reply", args) + } + + pub fn reconcile_reply( + &self, + group_id: &str, + reply_to: &str, + text: &str, + message_mode: ReplyMessageMode, + recipients: &[&str], + client_id: &str, + ) -> Result { + self.reply_idempotent( + group_id, + reply_to, + text, + message_mode, + recipients, + client_id, + ) + } + + /// Inspect unread Mail without moving the daemon-owned cursor. + pub fn inbox_peek(&self, group_id: &str, actor_id: &str, limit: u32) -> Result { + self.call_typed( + "inbox_peek", + object([ + ("group_id", Value::String(group_id.into())), + ("actor_id", Value::String(actor_id.into())), + ("limit", Value::from(limit)), + ]), + ) + } + + /// Atomically return and consume the next unread Mail prefix. + pub fn inbox_read(&self, group_id: &str, actor_id: &str, limit: u32) -> Result { + self.call_typed( + "inbox_read", + object([ + ("group_id", Value::String(group_id.into())), + ("actor_id", Value::String(actor_id.into())), + ("limit", Value::from(limit)), + ]), + ) + } +} + +fn insert_recipients(args: &mut Map, recipients: &[&str]) { + if !recipients.is_empty() { + args.insert( + "to".into(), + Value::Array( + recipients + .iter() + .map(|value| Value::String((*value).into())) + .collect(), + ), + ); + } +} + +fn deserialize_nullable_string<'de, D>(deserializer: D) -> std::result::Result +where + D: Deserializer<'de>, +{ + Ok(Option::::deserialize(deserializer)?.unwrap_or_default()) +} + +fn validate_client_id(client_id: &str) -> Result<()> { + if client_id.trim().is_empty() || client_id.len() > 256 { + return Err(Error::InvalidArgument( + "client_id must contain 1..=256 bytes".into(), + )); + } + Ok(()) +} + +fn validate_recipients(recipients: &[&str]) -> Result<()> { + if recipients + .iter() + .any(|recipient| recipient.trim().is_empty()) + { + return Err(Error::InvalidArgument( + "message recipients must be non-empty".into(), + )); + } + Ok(()) +} + +fn object(entries: [(&str, Value); N]) -> Map { + entries + .into_iter() + .map(|(key, value)| (key.to_owned(), value)) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::WorkloadIdentityEvidence; + use serde_json::json; + use std::io::{BufRead, BufReader, Write}; + use std::net::TcpListener; + use std::sync::{Arc, Mutex}; + use std::thread; + + #[derive(Clone)] + struct SignedIdentity(&'static str); + + impl WorkloadIdentityHook for SignedIdentity { + fn principal(&self) -> Result { + AuthenticatedPrincipal::new(self.0, "test-spiffe") + } + + fn evidence( + &self, + operation: &str, + _args: &Map, + ) -> Result { + WorkloadIdentityEvidence::new( + "workload_identity", + json!({"scheme": "test", "signature": format!("sig:{operation}")}), + ) + } + } + + fn server( + responses: Vec<&'static str>, + ) -> ( + crate::DaemonEndpoint, + Arc>>, + thread::JoinHandle<()>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let address = listener.local_addr().expect("address"); + let requests = Arc::new(Mutex::new(Vec::new())); + let captured = Arc::clone(&requests); + let handle = thread::spawn(move || { + for response in responses { + let (mut stream, _) = listener.accept().expect("accept"); + let mut request = String::new(); + BufReader::new(&mut stream) + .read_line(&mut request) + .expect("request"); + captured + .lock() + .expect("lock") + .push(serde_json::from_str(&request).expect("JSON request")); + stream.write_all(response.as_bytes()).expect("response"); + } + }); + ( + crate::DaemonEndpoint::Tcp { + host: "127.0.0.1".into(), + port: address.port(), + }, + requests, + handle, + ) + } + + fn message_response(id: &str, mode: &str, duplicate: bool) -> String { + serde_json::json!({ + "v": 1, + "ok": true, + "result": { + "event": { + "id": id, + "ts": "2026-08-29T01:00:00Z", + "kind": "chat.message", + "group_id": "g1", + "by": "workload:aquant", + "data": {"text": "hello", "message_mode": mode} + }, + "message_mode": mode, + "duplicate": duplicate + } + }) + .to_string() + + "\n" + } + + #[test] + fn maps_current_modes_and_reconciles_with_client_id() { + let responses = vec![ + Box::leak(message_response("e1", "mail", false).into_boxed_str()) as &'static str, + Box::leak(message_response("e1", "mail", true).into_boxed_str()), + Box::leak(message_response("e2", "send", false).into_boxed_str()), + Box::leak(message_response("e2", "send", true).into_boxed_str()), + ]; + let (endpoint, requests, handle) = server(responses); + let client = + IdentityBoundClient::new(CCCCClient::new(endpoint), SignedIdentity("workload:aquant")) + .expect("client"); + + let first = client + .send_idempotent("g1", "hello", MessageMode::Mail, &["peer1"], "send-1") + .expect("send"); + let replay = client + .reconcile_send("g1", "hello", MessageMode::Mail, &["peer1"], "send-1") + .expect("replay send"); + assert_eq!(first.event.id, replay.event.id); + assert!(replay.replayed); + + let first_reply = client + .reply_idempotent( + "g1", + "e1", + "done", + ReplyMessageMode::Send, + &["user"], + "reply-1", + ) + .expect("reply"); + let replay_reply = client + .reconcile_reply( + "g1", + "e1", + "done", + ReplyMessageMode::Send, + &["user"], + "reply-1", + ) + .expect("replay reply"); + assert_eq!(first_reply.event.id, replay_reply.event.id); + assert!(replay_reply.replayed); + + handle.join().expect("server"); + let requests = requests.lock().expect("requests"); + assert_eq!(requests[0]["op"], "send"); + assert_eq!(requests[0]["args"]["message_mode"], "mail"); + assert_eq!(requests[0]["args"]["client_id"], "send-1"); + assert_eq!(requests[0]["args"]["to"], json!(["peer1"])); + assert_eq!(requests[0]["args"]["by"], "workload:aquant"); + assert_eq!( + requests[0]["args"]["workload_identity"]["signature"], + "sig:send" + ); + assert_eq!(requests[2]["op"], "reply"); + assert_eq!(requests[2]["args"]["message_mode"], "send"); + } + + #[test] + fn uses_atomic_mail_inbox_operations() { + let peek = concat!( + r#"{"v":1,"ok":true,"result":{"messages":[{"id":"e1","ts":"now","kind":"chat.message","group_id":"g1","by":"user","data":{"message_mode":"mail"}}],"cursor":{"event_id":"","ts":""}}}"#, + "\n" + ); + let read = concat!( + r#"{"v":1,"ok":true,"result":{"messages":[{"id":"e1","ts":"now","kind":"chat.message","group_id":"g1","by":"user","data":{"message_mode":"mail"}}],"cursor":{"event_id":"e1","ts":"now","updated_at":"now"},"event":{"id":"r1","ts":"now","kind":"mail.read","group_id":"g1","by":"peer1","data":{"event_id":"e1"}}}}"#, + "\n" + ); + let (endpoint, requests, handle) = server(vec![peek, read]); + let client = IdentityBoundClient::new(CCCCClient::new(endpoint), SignedIdentity("peer1")) + .expect("client"); + + assert_eq!( + client + .inbox_peek("g1", "peer1", 5) + .expect("peek") + .messages + .len(), + 1 + ); + let consumed = client.inbox_read("g1", "peer1", 5).expect("read"); + assert_eq!(consumed.messages.len(), 1); + assert_eq!(consumed.cursor.event_id, "e1"); + assert_eq!(consumed.event.expect("read event").kind, "mail.read"); + + handle.join().expect("server"); + let requests = requests.lock().expect("requests"); + assert_eq!(requests[0]["op"], "inbox_peek"); + assert_eq!(requests[1]["op"], "inbox_read"); + assert_eq!(requests[1]["args"]["by"], "peer1"); + } + + #[test] + fn rejects_unstable_keys_and_blank_recipients_before_connecting() { + let client = IdentityBoundClient::new( + CCCCClient::new(crate::DaemonEndpoint::Tcp { + host: "127.0.0.1".into(), + port: 1, + }), + SignedIdentity("peer1"), + ) + .expect("client"); + + assert!(matches!( + client.send_idempotent("g1", "x", MessageMode::Send, &[], ""), + Err(Error::InvalidArgument(_)) + )); + assert!(matches!( + client.send_idempotent("g1", "x", MessageMode::Send, &[" "], "key"), + Err(Error::InvalidArgument(_)) + )); + } +} diff --git a/rust/tests/live_reliability.rs b/rust/tests/live_reliability.rs new file mode 100644 index 0000000..8a61619 --- /dev/null +++ b/rust/tests/live_reliability.rs @@ -0,0 +1,153 @@ +use cccc_sdk::{ + AuthenticatedPrincipal, CCCCClient, IdentityBoundClient, MessageMode, ReplyMessageMode, Result, + WorkloadIdentityEvidence, WorkloadIdentityHook, +}; +use serde_json::{json, Map, Value}; + +struct LiveTestIdentity(&'static str); + +impl WorkloadIdentityHook for LiveTestIdentity { + fn principal(&self) -> Result { + AuthenticatedPrincipal::new(self.0, "cccc-sdk-live-test") + } + + fn evidence( + &self, + operation: &str, + _args: &Map, + ) -> Result { + WorkloadIdentityEvidence::new( + "workload_identity", + json!({"test_only": true, "operation": operation}), + ) + } +} + +fn object(entries: impl IntoIterator) -> Map { + entries + .into_iter() + .map(|(key, value)| (key.to_owned(), value)) + .collect() +} + +#[test] +fn current_daemon_replays_writes_and_consumes_only_mail() { + if std::env::var_os("CCCC_RUN_LIVE_RELIABILITY").is_none() { + return; + } + + let raw = CCCCClient::discover().expect("discover live daemon"); + let created = raw + .call( + "group_create", + object([ + ("title", json!("cccc-sdk live reliability test")), + ("by", json!("user")), + ]), + ) + .expect("create disposable group"); + let group_id = created["group_id"].as_str().expect("group_id").to_owned(); + + let result = (|| { + raw.call( + "group_start", + object([("group_id", json!(group_id)), ("by", json!("user"))]), + )?; + raw.call( + "actor_add", + object([ + ("group_id", json!(group_id)), + ("actor_id", json!("sdk-live")), + ("runtime", json!("custom")), + ("runner", json!("headless")), + ("command", json!(["/usr/bin/true"])), + ("by", json!("user")), + ]), + )?; + + let sender = IdentityBoundClient::new(raw.clone(), LiveTestIdentity("user"))?; + let recipient = IdentityBoundClient::new(raw.clone(), LiveTestIdentity("sdk-live"))?; + + let send_key = format!("cccc-sdk-live:{group_id}:mail"); + let first = sender.send_idempotent( + &group_id, + "live Mail idempotency probe", + MessageMode::Mail, + &["sdk-live"], + &send_key, + )?; + let replay = sender.reconcile_send( + &group_id, + "live Mail idempotency probe", + MessageMode::Mail, + &["sdk-live"], + &send_key, + )?; + if first.event.id != replay.event.id || !replay.replayed { + return Err(cccc_sdk::Error::ReconciliationRequired( + "stable Mail client_id did not replay the original event".into(), + )); + } + + let peeked = recipient.inbox_peek(&group_id, "sdk-live", 10)?; + if peeked + .messages + .iter() + .all(|event| event.id != first.event.id) + { + return Err(cccc_sdk::Error::ReconciliationRequired( + "Mail event was absent from inbox_peek".into(), + )); + } + let consumed = recipient.inbox_read(&group_id, "sdk-live", 10)?; + if consumed + .messages + .iter() + .all(|event| event.id != first.event.id) + { + return Err(cccc_sdk::Error::ReconciliationRequired( + "atomic inbox_read did not return the Mail event".into(), + )); + } + if !recipient + .inbox_read(&group_id, "sdk-live", 10)? + .messages + .is_empty() + { + return Err(cccc_sdk::Error::ReconciliationRequired( + "Mail event remained unread after atomic inbox_read".into(), + )); + } + + let reply_key = format!("cccc-sdk-live:{group_id}:reply"); + let reply = recipient.reply_idempotent( + &group_id, + &first.event.id, + "live reply idempotency probe", + ReplyMessageMode::Send, + &["user"], + &reply_key, + )?; + let replay_reply = recipient.reconcile_reply( + &group_id, + &first.event.id, + "live reply idempotency probe", + ReplyMessageMode::Send, + &["user"], + &reply_key, + )?; + if reply.event.id != replay_reply.event.id || !replay_reply.replayed { + return Err(cccc_sdk::Error::ReconciliationRequired( + "stable reply client_id did not replay the original event".into(), + )); + } + Ok::<(), cccc_sdk::Error>(()) + })(); + + let cleanup = raw.call( + "group_delete", + object([("group_id", json!(group_id)), ("by", json!("user"))]), + ); + result.expect("live reliability assertions"); + cleanup.expect("delete disposable group"); +} diff --git a/scripts/check_sdk_hardening.py b/scripts/check_sdk_hardening.py new file mode 100644 index 0000000..499fafe --- /dev/null +++ b/scripts/check_sdk_hardening.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import ast +import re +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def read(relative: str) -> str: + return (ROOT / relative).read_text(encoding="utf-8") + + +def python_method(source: str, name: str) -> ast.FunctionDef | None: + tree = ast.parse(source) + return next( + (node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef) and node.name == name), + None, + ) + + +def main() -> int: + errors: list[str] = [] + + # Web Model completion remains an idempotent write. Keep its caller-stable + # replay key required even though the daemon can synthesize a default. + required = {"group_id", "actor_id", "turn_id", "delivery_id"} + python_source = read("python/src/cccc_sdk/client_0430_runtime_ops.py") + method = python_method(python_source, "web_model_runtime_complete_turn") + if method is None: + errors.append("Python completion wrapper is missing") + else: + parameters = { + arg.arg + for arg in (*method.args.posonlyargs, *method.args.args, *method.args.kwonlyargs) + } + if not required.issubset(parameters): + errors.append("Python completion wrapper does not require delivery_id") + if not any( + isinstance(node, ast.Dict) + and any(isinstance(key, ast.Constant) and key.value == "delivery_id" for key in node.keys) + for node in ast.walk(method) + ): + errors.append("Python completion wrapper does not map delivery_id") + + ts_types = read("ts/src/types.ts") + ts_runtime = read("ts/src/client_0430_runtime_ops.ts") + if not re.search(r"\bdeliveryId\s*:\s*string\s*;", ts_types): + errors.append("TypeScript deliveryId is missing or optional") + if not re.search(r"delivery_id\s*:\s*options\.deliveryId\b", ts_runtime): + errors.append("TypeScript completion wrapper does not map deliveryId") + + transport = read("ts/src/transport.ts") + transport_markers = { + "remainingTimeout(deadline)": "TypeScript transport does not share one deadline", + "signal?.removeEventListener('abort', onAbort)": "TypeScript abort cleanup is missing", + "assertBufferedLineLimit(remaining)": "TypeScript stream remainder is not byte-capped", + "initialBuffer: Buffer": "TypeScript stream buffering is not byte-accurate", + "OutcomeUnknownError": "TypeScript exchange failures lack an outcome-unknown boundary", + } + for marker, message in transport_markers.items(): + if marker not in transport: + errors.append(message) + if "removeAllListeners" in transport: + errors.append("TypeScript transport still removes unrelated socket listeners") + + reliable = read("rust/src/reliable.rs") + required_reliable_markers = { + "MessageMode": "Rust reliable send does not require an explicit message mode", + "ReplyMessageMode": "Rust reliable reply does not constrain reply modes", + '"client_id"': "Rust reliable writes do not carry a stable client_id", + '"inbox_peek"': "Rust reliable adapter lacks non-consuming Mail inspection", + '"inbox_read"': "Rust reliable adapter lacks atomic Mail consumption", + 'alias = "duplicate"': "Rust does not expose daemon duplicate replay state", + } + for marker, message in required_reliable_markers.items(): + if marker not in reliable: + errors.append(message) + + retired_markers = { + '"inbox_list"': "retired inbox_list leaked into the Rust adapter", + '"inbox_mark_read"': "retired inbox_mark_read leaked into the Rust adapter", + '"message_read_status"': "legacy per-message read status leaked into the Rust adapter", + "FileCursorStore": "a competing local Mail cursor remains in the Rust adapter", + "PersistentInbox": "legacy persistent inbox emulation remains in the Rust adapter", + } + for marker, message in retired_markers.items(): + if marker in reliable: + errors.append(message) + + workflows = { + name: read(f".github/workflows/{name}") + for name in ("python-integration.yml", "ts-ci.yml", "rust-ci.yml") + } + for name, workflow in workflows.items(): + if "repository: ChesterRa/cccc" not in workflow: + errors.append(f"{name} does not test against the current CCCC repository") + if "cargo install cccc --version '=0.4.33'" in workflow: + errors.append(f"{name} still pins the retired pre-message-cut daemon") + + rust_ci = workflows["rust-ci.yml"] + for marker, message in ( + ('toolchain: "1.74.0"', "Rust CI does not enforce the declared 1.74 MSRV"), + ("runs-on: windows-latest", "Rust CI does not test Windows"), + ('CCCC_RUN_LIVE_RELIABILITY: "1"', "Rust CI does not run live reliability checks"), + ("--test live_reliability", "Rust CI does not run the live reliability test"), + ): + if marker not in rust_ci: + errors.append(message) + + if errors: + for error in errors: + print(f"ERROR: {error}") + return 1 + + print("SDK hardening contract OK: replay keys, transport safety, atomic Mail, current native CI") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check_specs_against_cccc.sh b/scripts/check_specs_against_cccc.sh index f911332..f98a742 100755 --- a/scripts/check_specs_against_cccc.sh +++ b/scripts/check_specs_against_cccc.sh @@ -2,18 +2,31 @@ set -euo pipefail CCCC_REPO="${1:-../cccc}" +CCCC_REF="${2:-}" SRC="${CCCC_REPO%/}/docs/standards" ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" DST="${ROOT}/spec" -if [[ ! -d "${SRC}" ]]; then +if [[ -n "${CCCC_REF}" ]] && ! git -C "${CCCC_REPO}" rev-parse --verify "${CCCC_REF}^{commit}" >/dev/null 2>&1; then + echo "error: unknown CCCC git ref: ${CCCC_REF}" >&2 + exit 2 +fi + +if [[ -z "${CCCC_REF}" && ! -d "${SRC}" ]]; then echo "error: cannot find CCCC specs at: ${SRC}" >&2 exit 2 fi status=0 for name in CCCS_V1.md CCCC_DAEMON_IPC_V1.md CCCC_CONTEXT_OPS_V1.md; do - if ! cmp -s "${SRC}/${name}" "${DST}/${name}"; then + if [[ -n "${CCCC_REF}" ]]; then + if git -C "${CCCC_REPO}" show "${CCCC_REF}:docs/standards/${name}" | cmp -s - "${DST}/${name}"; then + continue + fi + echo "error: spec/${name} has drifted from CCCC core" >&2 + diff -u "${DST}/${name}" <(git -C "${CCCC_REPO}" show "${CCCC_REF}:docs/standards/${name}") || true + status=1 + elif ! cmp -s "${SRC}/${name}" "${DST}/${name}"; then echo "error: spec/${name} has drifted from CCCC core" >&2 diff -u "${DST}/${name}" "${SRC}/${name}" || true status=1 @@ -24,4 +37,8 @@ if [[ "${status}" -ne 0 ]]; then exit "${status}" fi -echo "All mirrored CCCC standards match core." +if [[ -n "${CCCC_REF}" ]]; then + echo "All mirrored CCCC standards match ${CCCC_REPO}@${CCCC_REF}." +else + echo "All mirrored CCCC standards match core." +fi diff --git a/scripts/sync_specs_from_cccc.sh b/scripts/sync_specs_from_cccc.sh index d83411e..1cffd6c 100755 --- a/scripts/sync_specs_from_cccc.sh +++ b/scripts/sync_specs_from_cccc.sh @@ -2,13 +2,38 @@ set -euo pipefail CCCC_REPO="${1:-../cccc}" -SRC="${CCCC_REPO%/}/docs/standards" +CCCC_REF="${2:-}" ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" DST="${ROOT}/spec" +TMP="" + +cleanup() { + if [[ -n "${TMP}" && -d "${TMP}" ]]; then + rm -rf -- "${TMP}" + fi +} +trap cleanup EXIT + +if [[ -n "${CCCC_REF}" ]]; then + if ! git -C "${CCCC_REPO}" rev-parse --verify "${CCCC_REF}^{commit}" >/dev/null 2>&1; then + echo "error: unknown CCCC git ref: ${CCCC_REF}" >&2 + exit 2 + fi + TMP="$(mktemp -d)" + git -C "${CCCC_REPO}" archive "${CCCC_REF}" \ + docs/standards/CCCS_V1.md \ + docs/standards/CCCC_DAEMON_IPC_V1.md \ + docs/standards/CCCC_CONTEXT_OPS_V1.md | tar -x -C "${TMP}" + SRC="${TMP}/docs/standards" + SOURCE_LABEL="${CCCC_REPO}@${CCCC_REF}" +else + SRC="${CCCC_REPO%/}/docs/standards" + SOURCE_LABEL="${SRC}" +fi if [[ ! -d "${SRC}" ]]; then echo "error: cannot find CCCC specs at: ${SRC}" >&2 - echo "hint: pass the CCCC repo path explicitly: ./scripts/sync_specs_from_cccc.sh /path/to/cccc" >&2 + echo "hint: ./scripts/sync_specs_from_cccc.sh /path/to/cccc [git-ref]" >&2 exit 2 fi @@ -17,5 +42,4 @@ cp -f "${SRC}/CCCS_V1.md" "${DST}/CCCS_V1.md" cp -f "${SRC}/CCCC_DAEMON_IPC_V1.md" "${DST}/CCCC_DAEMON_IPC_V1.md" cp -f "${SRC}/CCCC_CONTEXT_OPS_V1.md" "${DST}/CCCC_CONTEXT_OPS_V1.md" -echo "Synced specs from ${SRC} -> ${DST}" - +echo "Synced specs from ${SOURCE_LABEL} -> ${DST}" diff --git a/spec/ADAPTATION_PLAN.md b/spec/ADAPTATION_PLAN.md index 61412ce..d7d8129 100644 --- a/spec/ADAPTATION_PLAN.md +++ b/spec/ADAPTATION_PLAN.md @@ -2,73 +2,79 @@ Updated 2026-08-29 for the CCCC 0.4.36 native-only transition. This document describes source compatibility work; package version changes and publication -are separate release decisions. +remain separate release decisions. ## Product boundary -- CCCC ships one native Rust daemon/web/CLI product and owns all runtime state - under `CCCC_HOME`. -- This repository ships Python, TypeScript, and Rust **client** SDKs. An SDK's +- CCCC ships one native Rust daemon/web/CLI product and owns runtime state under + `CCCC_HOME`. +- This repository ships Python, TypeScript, and Rust client SDKs. An SDK's implementation language does not select, embed, or replace the daemon. -- The three files mirrored from `cccc/docs/standards/` are authoritative. SDK - helpers may improve language ergonomics but must preserve their wire shapes. +- The three files mirrored from `cccc/docs/standards/` are authoritative. - Compatibility is determined by `ipc_v`, capabilities, and safe operation - probes. The bundled daemon reports `implementation="rust"`, but clients do - not treat that label alone as a compatibility proof. + probes, not by exact product-version or implementation-label equality. ## Current public alignment The 0.4.36 source target makes one atomic messaging cut: - every new message selects `send`, `request_reply`, or `mail`; -- one audience domain is allowed per message: human user or agents, never both; +- one message addresses the human user or agents, never both audience domains; - Mail is agent-only, enters the Mail Inbox, and does not immediately prompt a runtime; -- replies select `send` (default) or `mail`; both fulfill the original reply - request, and replies cannot create another generic reply request; -- Inbox peek/read, non-consuming message history, manual delivery, and - reply-request cancellation use their current daemon operations; -- retired generic ACK/read operations and legacy delivery fields are not - translated or silently downgraded. +- replies select `send` or `mail`; both fulfill the original reply request; +- `inbox_peek` and atomic `inbox_read` replace the retired generic ACK/read + model; +- non-consuming history, manual delivery, and reply-request cancellation use + their current daemon operations. -Context and Group Space helpers follow the same current contract: +The Rust identity-bound adapter follows the same boundary. Stable `client_id` +values reconcile ambiguous send/reply writes, while Mail consumption delegates +to the daemon's atomic `inbox_read` transaction. It does not emulate removed +`inbox_list`, `inbox_mark_read`, or per-message ACK operations. + +Transport behavior follows the normative safety rules: requests are bounded +before connecting, response IPC versions are validated, auto-discovered +endpoints refresh only after a pre-write connection failure, and failures after +exchange begins are reported as outcome-unknown without automatic replay. + +Context and Group Space helpers follow the current contract: - `context_get` exposes `overview`, `summary`, and `full` projections; -- existing task-list helpers support exact lookup/batches, filters, atomic - status pages, and pagination; -- `group_space_sync` is legacy read-only status. Explicit ingest/source - operations are the mutation path. +- task-list helpers support exact lookup/batches, filters, atomic status pages, + and pagination; +- `group_space_sync` is legacy read-only status; explicit ingest/source + operations remain the mutation path. ## Intentional generic-only operations Public SDK quality is not measured by wrapper count. These boundaries remain -generic unless an external consumer requires a stable typed API: +generic until a concrete external consumer requires a stable typed API: - `message_upload_preflight`, which coordinates Web-owned temporary uploads; - internal Group Bridge relay/record operations; -- duplex browser, terminal, and attachment upgrade operations that require a - dedicated streaming ownership model; +- duplex browser, terminal, and attachment upgrades that require dedicated + stream ownership and backpressure semantics; - administrative or provider internals without a portable external workflow. All non-streaming operations remain reachable through `call` / `call_raw`. -Compatibility probing must skip operations whose probe would mutate state or -open an upgraded stream. +Compatibility probing skips operations that mutate state or open an upgraded +stream. ## Source and release gates Before committing an alignment change: -1. `scripts/check_specs_against_cccc.sh` must prove that all mirrored standards - are byte-identical to the selected CCCC checkout. -2. Python, TypeScript, and Rust contract/transport suites must pass. +1. `scripts/check_specs_against_cccc.sh` proves all mirrored standards are + byte-identical to the selected committed CCCC revision. +2. Python, TypeScript, and Rust contract/transport suites pass. 3. Source compilation/type checking, package builds, Rust formatting, clippy, - and packaging checks must pass. -4. Examples and READMEs must describe one native daemon and three independent - client languages; obsolete dual-engine or `ccccd` command guidance must not - remain. -5. Package manifests stay unchanged until the release version is explicitly - chosen. No sync task may tag, publish, or deploy artifacts. + MSRV, and packaging checks pass. +4. Current-daemon integration covers the atomic message modes and Mail Inbox. +5. Examples and READMEs describe one native daemon and three client languages; + obsolete dual-engine, ACK/read, and `ccccd` guidance is absent. +6. No sync task tags, publishes, or deploys artifacts. Rollback is repository-local: revert the SDK commit without changing the CCCC -runtime or `CCCC_HOME`. Because this repository contains clients only, contract -alignment must never mutate daemon state during build or test. +runtime or `CCCC_HOME`. Contract alignment must never mutate daemon state during +ordinary build or unit-test gates. diff --git a/ts/DESIGN.md b/ts/DESIGN.md index 9e9ee12..3810f82 100644 --- a/ts/DESIGN.md +++ b/ts/DESIGN.md @@ -25,6 +25,7 @@ Out of scope: - `src/transport.ts`: endpoint discovery, socket I/O, events stream handshake. - `src/client.ts`: high-level SDK methods. +- `src/client_0430_runtime_ops.ts`: retained Web Model wait/complete operations. - `src/client_0434_ops.ts`: current terminal/Web Model contract additions. - `src/types.ts`: IPC-facing option and payload types. - `src/errors.ts`: typed error hierarchy. @@ -43,6 +44,8 @@ Out of scope: - Connection-establishment failures -> `DaemonConnectionError` and one safe endpoint rediscovery for auto-discovered clients. - Failures after exchange begins -> `OutcomeUnknownError` and no automatic replay. +- Connection/response/handshake phases share one caller deadline; stream bytes + are capped before UTF-8 decoding. - Oversized requests -> `RequestTooLargeError` before connecting. - Daemon `ok:false` responses -> `DaemonAPIError` with `code/message/details/raw`. - Compatibility failures -> `IncompatibleDaemonError`. diff --git a/ts/README.md b/ts/README.md index 050e42b..21e8469 100644 --- a/ts/README.md +++ b/ts/README.md @@ -239,8 +239,27 @@ const recovered = await client.webModelRuntimeRecoverTurn({ actorId: 'web-model', eventIds: ['e_xxx'], }); + +// Complete a runtime-owned turn. Reuse the exact deliveryId after an unknown +// outcome so the daemon can replay the same completion receipt. +const acquired = await client.webModelRuntimeWaitNextTurn({ + groupId, + actorId: 'web-model', +}); +const turn = acquired.turn as { turn_id: string; event_ids: string[] }; +const deliveryId = `worker:${turn.turn_id}`; +await client.webModelRuntimeCompleteTurn({ + groupId, + actorId: 'web-model', + turnId: turn.turn_id, + deliveryId, + eventIds: turn.event_ids, +}); ``` +`deliveryId` is required and is the completion replay key. A retry for the +same acquired turn must reuse the same value. + `termResize()` sends the standard `term_resize` operation. For older compatible daemon builds that expose `terminal_resize`, the SDK falls back only after receiving a structured `unknown_op`; transport failures are never diff --git a/ts/__tests__/client_0430_contract.test.ts b/ts/__tests__/client_0430_contract.test.ts index 5760b75..4bf514e 100644 --- a/ts/__tests__/client_0430_contract.test.ts +++ b/ts/__tests__/client_0430_contract.test.ts @@ -16,7 +16,7 @@ async function makeClient(calls: CallCapture[]): Promise { return client; } -describe('cccc 0.4.33 JSON op alignment', () => { +describe('current native CCCC JSON op alignment', () => { it('maps current message, group preamble, and terminal operations', async () => { const calls: CallCapture[] = []; const client = await makeClient(calls); @@ -335,4 +335,27 @@ describe('cccc 0.4.33 JSON op alignment', () => { assert.equal(calls[1]?.args?.['text'], 'Include omissions'); assert.equal(calls[1]?.args?.['source_text'], 'Current meeting notes'); }); + + it('requires and reuses deliveryId for Web Model completion replay', async () => { + const calls: CallCapture[] = []; + const client = await makeClient(calls); + const options = { + groupId: 'g_1', + actorId: 'web-model', + turnId: 'turn-1', + deliveryId: 'worker:turn-1', + eventIds: ['e_1'], + status: 'done' as const, + }; + + await client.webModelRuntimeCompleteTurn(options); + await client.webModelRuntimeCompleteTurn(options); + + assert.deepEqual(calls[0], calls[1]); + assert.equal(calls[0]?.op, 'web_model_runtime_complete_turn'); + for (const required of ['group_id', 'actor_id', 'turn_id', 'delivery_id']) { + assert.ok(required in (calls[0]?.args ?? {}), `missing daemon arg: ${required}`); + } + assert.equal(calls[0]?.args?.['delivery_id'], options.deliveryId); + }); }); diff --git a/ts/__tests__/transport.test.ts b/ts/__tests__/transport.test.ts index caabf04..09e747c 100644 --- a/ts/__tests__/transport.test.ts +++ b/ts/__tests__/transport.test.ts @@ -6,6 +6,7 @@ import * as fs from 'node:fs/promises'; import * as net from 'node:net'; import { Readable } from 'node:stream'; import { + callDaemon, discoverEndpoint, defaultHome, openEventsStream, @@ -13,6 +14,52 @@ import { MAX_LINE_SIZE, DEFAULT_TIMEOUT_MS, } from '../src/transport.js'; +import type { DaemonEndpoint, DaemonRequest } from '../src/types.js'; + +interface TestServer { + endpoint: DaemonEndpoint; + sockets: Set; + close(): Promise; +} + +async function startServer(onConnection: (socket: net.Socket) => void): Promise { + const sockets = new Set(); + const server = net.createServer((socket) => { + sockets.add(socket); + socket.once('close', () => sockets.delete(socket)); + onConnection(socket); + socket.resume(); + }); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + if (address === null || typeof address === 'string') { + throw new Error('test server did not bind a TCP port'); + } + return { + endpoint: { + transport: 'tcp', + host: '127.0.0.1', + port: address.port, + path: '', + }, + sockets, + close: async () => { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + }); + }, + }; +} + +const streamRequest: DaemonRequest = { + v: 1, + op: 'events_stream', + args: { group_id: 'g1', by: 'user' }, +}; describe('defaultHome', () => { it('returns CCCC_HOME env if set', () => { @@ -251,6 +298,61 @@ describe('openEventsStream abort handling', () => { }); }); +describe('daemon response deadlines', () => { + it('times out after TCP connect when the daemon never responds', async () => { + const server = await startServer(() => undefined); + const started = Date.now(); + try { + await assert.rejects( + callDaemon(server.endpoint, { v: 1, op: 'ping', args: {} }, 50), + /Response timeout/, + ); + assert.ok(Date.now() - started < 1_000, 'response timeout should be bounded'); + } finally { + await server.close(); + } + }); +}); + +describe('event stream handshake safety', () => { + it('rejects a daemon that accepts the socket but never handshakes', async () => { + const server = await startServer(() => undefined); + const started = Date.now(); + try { + await assert.rejects( + openEventsStream(server.endpoint, streamRequest, 50), + /Handshake timeout/, + ); + assert.ok(Date.now() - started < 1_000, 'handshake timeout should be bounded'); + } finally { + await server.close(); + } + }); + + it('honors AbortSignal while waiting for the handshake', async () => { + const server = await startServer(() => undefined); + const controller = new AbortController(); + try { + const pending = openEventsStream(server.endpoint, streamRequest, 5_000, controller.signal); + setTimeout(() => controller.abort(), 20); + await assert.rejects(pending, /aborted/); + const closeDeadline = Date.now() + 1_000; + while (server.sockets.size !== 0 && Date.now() < closeDeadline) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.equal(server.sockets.size, 0, 'aborting the handshake must close the socket'); + } finally { + await server.close(); + } + }); + + it('applies the byte cap to data buffered with the handshake', async () => { + const socket = Readable.from([]) as unknown as net.Socket; + const lines = readLines(socket, Buffer.alloc(MAX_LINE_SIZE + 1, 0x78)); + await assert.rejects(lines.next(), /Stream line exceeds MAX_LINE_SIZE/); + }); +}); + describe('readLines', () => { it('preserves UTF-8 code points split across socket chunks', async () => { const encoded = Buffer.from('{"text":"中文"}\n', 'utf8'); diff --git a/ts/src/client_0430_ops.ts b/ts/src/client_0430_ops.ts index 23f883c..a961278 100644 --- a/ts/src/client_0430_ops.ts +++ b/ts/src/client_0430_ops.ts @@ -1,12 +1,15 @@ import { installCCCC0430AdminOps, type CCCC0430AdminOps } from './client_0430_admin_ops.js'; import { installCCCC0430AssistantOps, type CCCC0430AssistantOps } from './client_0430_assistant_ops.js'; import { installCCCC0430MemoryOps, type CCCC0430MemoryOps } from './client_0430_memory_ops.js'; +import { installCCCC0430RuntimeOps, type CCCC0430RuntimeOps } from './client_0430_runtime_ops.js'; import type { CCCC0430Client } from './client_0430_shared.js'; -export interface CCCC0430Ops extends CCCC0430AdminOps, CCCC0430AssistantOps, CCCC0430MemoryOps {} +export interface CCCC0430Ops + extends CCCC0430AdminOps, CCCC0430AssistantOps, CCCC0430MemoryOps, CCCC0430RuntimeOps {} export function installCCCC0430Ops(proto: CCCC0430Client & Partial): void { installCCCC0430AdminOps(proto); installCCCC0430AssistantOps(proto); installCCCC0430MemoryOps(proto); + installCCCC0430RuntimeOps(proto); } diff --git a/ts/src/client_0430_runtime_ops.ts b/ts/src/client_0430_runtime_ops.ts new file mode 100644 index 0000000..c9e85f2 --- /dev/null +++ b/ts/src/client_0430_runtime_ops.ts @@ -0,0 +1,46 @@ +import { compactRecord, type CCCC0430Client } from './client_0430_shared.js'; +import type { + WebModelRuntimeCompleteTurnOptions, + WebModelRuntimeWaitNextTurnOptions, +} from './types.js'; + +export interface CCCC0430RuntimeOps { + webModelRuntimeWaitNextTurn( + options: WebModelRuntimeWaitNextTurnOptions + ): Promise>; + webModelRuntimeCompleteTurn( + options: WebModelRuntimeCompleteTurnOptions + ): Promise>; +} + +const runtimeOps: CCCC0430RuntimeOps & ThisType = { + async webModelRuntimeWaitNextTurn(options) { + return this.call('web_model_runtime_wait_next_turn', { + group_id: options.groupId, + actor_id: options.actorId, + by: options.by ?? options.actorId, + limit: Math.min(Math.max(Math.trunc(options.limit ?? 20), 1), 20), + kind_filter: options.kindFilter ?? 'all', + }); + }, + + async webModelRuntimeCompleteTurn(options) { + return this.call('web_model_runtime_complete_turn', compactRecord({ + group_id: options.groupId, + actor_id: options.actorId, + by: options.by ?? options.actorId, + turn_id: options.turnId, + delivery_id: options.deliveryId, + event_ids: options.eventIds, + latest_event_id: options.latestEventId, + status: options.status ?? 'done', + summary: options.summary, + })); + }, +}; + +export function installCCCC0430RuntimeOps( + proto: CCCC0430Client & Partial +): void { + Object.assign(proto, runtimeOps); +} diff --git a/ts/src/index.ts b/ts/src/index.ts index 0231ea0..aab4784 100644 --- a/ts/src/index.ts +++ b/ts/src/index.ts @@ -157,6 +157,9 @@ export type { TerminalSinceOptions, TerminalSnapshotOptions, TerminalClearOptions, + WebModelRuntimeWaitNextTurnOptions, + WebModelRuntimeCompletionStatus, + WebModelRuntimeCompleteTurnOptions, WebModelDeliveryMode, WebModelDeliveryPreferencesGetOptions, WebModelDeliveryPreferencesUpdateOptions, diff --git a/ts/src/transport.ts b/ts/src/transport.ts index cdd6b8d..1b9c57d 100644 --- a/ts/src/transport.ts +++ b/ts/src/transport.ts @@ -121,9 +121,12 @@ function connect( return new Promise((resolve, reject) => { const socket = new net.Socket(); let settled = false; + let timer: ReturnType | undefined; const cleanup = () => { - socket.removeAllListeners(); + if (timer !== undefined) clearTimeout(timer); + socket.removeListener('connect', onConnect); + socket.removeListener('error', onError); signal?.removeEventListener('abort', onAbort); }; @@ -154,22 +157,58 @@ function connect( return; } signal?.addEventListener('abort', onAbort, { once: true }); - socket.setTimeout(timeoutMs); + timer = setTimeout(onTimeout, timeoutMs); + socket.once('connect', onConnect); socket.once('error', onError); - socket.once('timeout', onTimeout); - - if (endpoint.transport === 'tcp') { - socket.connect(endpoint.port, endpoint.host, onConnect); - } else if (endpoint.transport === 'unix') { - socket.connect(endpoint.path, onConnect); - } else { - rejectAndDestroy( - new DaemonConnectionError(`Invalid endpoint transport: ${endpoint.transport}`), - ); + + try { + if (endpoint.transport === 'tcp') { + socket.connect(endpoint.port, endpoint.host); + } else if (endpoint.transport === 'unix') { + socket.connect(endpoint.path); + } else { + rejectAndDestroy( + new DaemonConnectionError(`Invalid endpoint transport: ${endpoint.transport}`), + ); + } + } catch (error) { + rejectAndDestroy(new DaemonConnectionError( + error instanceof Error ? error.message : String(error), + )); } }); } +function remainingTimeout(deadline: number): number { + return Math.max(1, deadline - Date.now()); +} + +function appendChunk(buffer: Buffer, chunk: Buffer): Buffer { + return buffer.length === 0 ? chunk : Buffer.concat([buffer, chunk]); +} + +function assertBufferedLineLimit(buffer: Buffer): void { + let start = 0; + let newlineIndex: number; + while ((newlineIndex = buffer.indexOf(0x0a, start)) !== -1) { + if (newlineIndex - start > MAX_LINE_SIZE) { + throw new DaemonUnavailableError( + `Stream line exceeds MAX_LINE_SIZE (${MAX_LINE_SIZE} bytes)`, + ); + } + start = newlineIndex + 1; + } + if (buffer.length - start > MAX_LINE_SIZE) { + throw new DaemonUnavailableError( + `Stream line exceeds MAX_LINE_SIZE (${MAX_LINE_SIZE} bytes)`, + ); + } +} + +function decodeLine(bytes: Buffer): string { + return new TextDecoder('utf-8', { fatal: true }).decode(bytes); +} + // ============================================================ // IPC calls // ============================================================ @@ -189,6 +228,7 @@ export async function callDaemon( request: DaemonRequest, timeoutMs: number = DEFAULT_TIMEOUT_MS ): Promise { + const deadline = Date.now() + timeoutMs; const line = JSON.stringify(request) + '\n'; if (Buffer.byteLength(line, 'utf8') > MAX_REQUEST_SIZE) { throw new RequestTooLargeError(`Daemon request exceeds ${MAX_REQUEST_SIZE} bytes`); @@ -196,19 +236,31 @@ export async function callDaemon( const socket = await connect(endpoint, timeoutMs); return new Promise((resolve, reject) => { - let buffer = Buffer.alloc(0); - let resolved = false; + let buffer: Buffer = Buffer.alloc(0); + let settled = false; + let timer: ReturnType | undefined; const cleanup = () => { - socket.removeAllListeners(); + if (timer !== undefined) clearTimeout(timer); + socket.removeListener('data', onData); + socket.removeListener('error', onError); + socket.removeListener('close', onClose); }; - socket.on('data', (chunk: Buffer) => { - buffer = Buffer.concat([buffer, chunk]); + const fail = (error: Error) => { + if (settled) return; + settled = true; + cleanup(); + socket.destroy(); + reject(error); + }; + const onData = (chunk: Buffer) => { + if (settled) return; + buffer = appendChunk(buffer, chunk); const newlineIndex = buffer.indexOf(0x0a); - if (newlineIndex !== -1 && !resolved) { - resolved = true; + if (newlineIndex !== -1) { + settled = true; const responseBytes = buffer.subarray(0, newlineIndex); cleanup(); socket.destroy(); @@ -238,49 +290,34 @@ export async function callDaemon( } if (newlineIndex === -1 && buffer.length > MAX_LINE_SIZE) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Response too large')); + fail(new OutcomeUnknownError(request.op, 'Response too large')); } - }); - - socket.on('error', (err) => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, err.message)); - } - }); + }; - socket.on('close', () => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Connection closed unexpectedly')); - } - }); + const onError = (err: Error) => fail(new OutcomeUnknownError(request.op, err.message)); + const onClose = () => fail( + new OutcomeUnknownError(request.op, 'Connection closed unexpectedly'), + ); - socket.once('timeout', () => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Response timeout')); - } - }); + socket.on('data', onData); + socket.once('error', onError); + socket.once('close', onClose); + timer = setTimeout( + () => fail(new OutcomeUnknownError(request.op, 'Response timeout')), + remainingTimeout(deadline), + ); // Send request. - socket.write(line, (err) => { - if (err && !resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, `Write failed: ${err.message}`)); - } - }); + try { + socket.write(line, (err) => { + if (err) fail(new OutcomeUnknownError(request.op, `Write failed: ${err.message}`)); + }); + } catch (error) { + fail(new OutcomeUnknownError( + request.op, + error instanceof Error ? error.message : String(error), + )); + } }); } @@ -311,6 +348,7 @@ export async function openEventsStream( timeoutMs: number = DEFAULT_TIMEOUT_MS, signal?: AbortSignal, ): Promise { + const deadline = Date.now() + timeoutMs; if (signal?.aborted) { throw new DaemonUnavailableError('Event stream aborted'); } @@ -330,107 +368,95 @@ export async function openEventsStream( handshake: DaemonResponse; remainingBuffer: Buffer; }>((resolve, reject) => { - let buffer = Buffer.alloc(0); - let resolved = false; + let buffer: Buffer = Buffer.alloc(0); + let settled = false; + let timer: ReturnType | undefined; const cleanup = () => { - socket.removeAllListeners(); + if (timer !== undefined) clearTimeout(timer); + socket.removeListener('data', onData); + socket.removeListener('error', onError); + socket.removeListener('close', onClose); signal?.removeEventListener('abort', onAbort); }; - const onAbort = () => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new DaemonUnavailableError('Event stream aborted')); - } + const fail = (error: Error) => { + if (settled) return; + settled = true; + cleanup(); + socket.destroy(); + reject(error); }; + const onAbort = () => fail(new DaemonUnavailableError('Event stream aborted')); + const onData = (chunk: Buffer) => { - buffer = Buffer.concat([buffer, chunk]); + if (settled) return; + buffer = appendChunk(buffer, chunk); const newlineIndex = buffer.indexOf(0x0a); - if (newlineIndex !== -1 && !resolved) { - resolved = true; - cleanup(); + if (newlineIndex !== -1) { const responseBytes = buffer.subarray(0, newlineIndex); const remaining = buffer.subarray(newlineIndex + 1); if (responseBytes.length > MAX_LINE_SIZE) { - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Handshake response too large')); + fail(new OutcomeUnknownError(request.op, 'Handshake response too large')); return; } try { const parsed: unknown = JSON.parse(responseBytes.toString('utf8')); if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) { - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Handshake must be a JSON object')); + fail(new OutcomeUnknownError(request.op, 'Handshake must be a JSON object')); return; } const handshake = parsed as DaemonResponse; if (handshake.v !== 1) { - socket.destroy(); - reject(new IncompatibleDaemonError( + fail(new IncompatibleDaemonError( `Daemon stream handshake uses unsupported IPC version: ${String(handshake.v)}`, )); return; } + assertBufferedLineLimit(remaining); + settled = true; + cleanup(); resolve({ handshake, remainingBuffer: remaining, }); - } catch { - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Invalid handshake JSON')); + } catch (error) { + fail(error instanceof DaemonUnavailableError + ? error + : new OutcomeUnknownError(request.op, 'Invalid handshake JSON')); } } if (newlineIndex === -1 && buffer.length > MAX_LINE_SIZE) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Handshake response too large')); + fail(new OutcomeUnknownError(request.op, 'Handshake response too large')); } }; + const onError = (err: Error) => fail(new OutcomeUnknownError(request.op, err.message)); + const onClose = () => fail( + new OutcomeUnknownError(request.op, 'Connection closed during handshake'), + ); + socket.on('data', onData); signal?.addEventListener('abort', onAbort, { once: true }); - socket.once('error', (err) => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, err.message)); - } - }); - socket.once('close', () => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Connection closed during handshake')); - } - }); - socket.once('timeout', () => { - if (!resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, 'Handshake timeout')); - } - }); - socket.write(line, (error) => { - if (error && !resolved) { - resolved = true; - cleanup(); - socket.destroy(); - reject(new OutcomeUnknownError(request.op, `Write failed: ${error.message}`)); - } - }); + socket.once('error', onError); + socket.once('close', onClose); + timer = setTimeout( + () => fail(new OutcomeUnknownError(request.op, 'Handshake timeout')), + remainingTimeout(deadline), + ); + try { + socket.write(line, (error) => { + if (error) fail(new OutcomeUnknownError(request.op, `Write failed: ${error.message}`)); + }); + } catch (error) { + fail(new OutcomeUnknownError( + request.op, + error instanceof Error ? error.message : String(error), + )); + } }); - // Remove timeout after handshake. - socket.setTimeout(0); - return { socket, handshake, initialBuffer: remainingBuffer }; } @@ -445,53 +471,49 @@ export async function* readLines( socket: net.Socket, initialBuffer: string | Buffer = '' ): AsyncGenerator { - const decoder = new TextDecoder('utf-8', { fatal: true }); let buffer = typeof initialBuffer === 'string' - ? initialBuffer - : decoder.decode(initialBuffer, { stream: true }); + ? Buffer.from(initialBuffer, 'utf8') + : initialBuffer; - const ensureBounded = (line: string): void => { - if (Buffer.byteLength(line, 'utf8') > MAX_LINE_SIZE) { + // Handle lines from initial buffer. + let newlineIndex: number; + while ((newlineIndex = buffer.indexOf(0x0a)) !== -1) { + if (newlineIndex > MAX_LINE_SIZE) { throw new DaemonUnavailableError( `Stream line exceeds MAX_LINE_SIZE (${MAX_LINE_SIZE} bytes)`, ); } - }; - - // Handle lines from initial buffer. - let newlineIndex: number; - while ((newlineIndex = buffer.indexOf('\n')) !== -1) { - const line = buffer.slice(0, newlineIndex); - buffer = buffer.slice(newlineIndex + 1); - ensureBounded(line); + const line = decodeLine(buffer.subarray(0, newlineIndex)); + buffer = buffer.subarray(newlineIndex + 1); if (line.trim()) { yield line; } } + assertBufferedLineLimit(buffer); // Continue reading from socket. for await (const chunk of socket) { - buffer += decoder.decode(chunk as Buffer, { stream: true }); - - if (buffer.indexOf('\n') === -1 && Buffer.byteLength(buffer, 'utf8') > MAX_LINE_SIZE) { - throw new DaemonUnavailableError( - `Stream line exceeds MAX_LINE_SIZE (${MAX_LINE_SIZE} bytes)` - ); - } - - while ((newlineIndex = buffer.indexOf('\n')) !== -1) { - const line = buffer.slice(0, newlineIndex); - buffer = buffer.slice(newlineIndex + 1); - ensureBounded(line); + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array); + buffer = appendChunk(buffer, bytes); + + while ((newlineIndex = buffer.indexOf(0x0a)) !== -1) { + if (newlineIndex > MAX_LINE_SIZE) { + throw new DaemonUnavailableError( + `Stream line exceeds MAX_LINE_SIZE (${MAX_LINE_SIZE} bytes)`, + ); + } + const line = decodeLine(buffer.subarray(0, newlineIndex)); + buffer = buffer.subarray(newlineIndex + 1); if (line.trim()) { yield line; } } + assertBufferedLineLimit(buffer); } - buffer += decoder.decode(); - if (buffer.trim()) { - ensureBounded(buffer); - yield buffer; + if (buffer.length > 0) { + assertBufferedLineLimit(buffer); + const line = decodeLine(buffer); + if (line.trim()) yield line; } } diff --git a/ts/src/types.ts b/ts/src/types.ts index 96aadc7..b0b8b44 100644 --- a/ts/src/types.ts +++ b/ts/src/types.ts @@ -697,6 +697,34 @@ export interface TerminalSnapshotOptions { by?: string; } +/** Read the next coalesced turn for one Web Model runtime actor. */ +export interface WebModelRuntimeWaitNextTurnOptions { + groupId: string; + actorId: string; + by?: string; + limit?: number; + kindFilter?: 'all' | 'chat' | 'notify'; +} + +export type WebModelRuntimeCompletionStatus = + | 'done' + | 'partial' + | 'failed' + | 'cancelled'; + +/** Complete a previously acquired Web Model turn with a stable replay key. */ +export interface WebModelRuntimeCompleteTurnOptions { + groupId: string; + actorId: string; + turnId: string; + deliveryId: string; + eventIds?: string[]; + latestEventId?: string; + status?: WebModelRuntimeCompletionStatus; + summary?: string; + by?: string; +} + export type WebModelDeliveryMode = 'standard' | 'image_compat'; export interface WebModelDeliveryPreferencesGetOptions {