Merge pull request 'fix(pypi): swap OIDC trusted-publisher for twine + PYPI_TOKEN; port .github -> .gitea' (#6) from fix/pypi-gitea-twine-no-oidc into main

Post-2026-05-06 PyPI Trusted-Publisher OIDC is dead for our repos — PyPI
only accepts GitHub/GitLab/Google/ActiveState issuers, not Gitea. This PR:

1. Renames .github/workflows/{ci,publish}.yml -> .gitea/workflows/. (Gitea
   Actions reads .gitea/ exclusively on this repo; the .github/ path was
   silently dead since the migration — saved memory
   reference_molecule_core_actions_gitea_only.)

2. Replaces `pypa/gh-action-pypi-publish` (which requires OIDC id-token
   exchange that PyPI rejects from Gitea) with `python -m twine upload
   --username __token__ --password "$PYPI_TOKEN"`. Mirrors the canonical
   pattern in molecule-core/.gitea/workflows/publish-runtime.yml that has
   been shipping successfully since 2026-05-11.

3. Drops `permissions: id-token: write` (no longer needed without OIDC).

4. Adds `twine check` to the build step (catches metadata regressions
   before upload).

5. Adds concurrency group to serialize tag-driven publishes.

6. Updates README "Releasing" section to describe the twine+SSOT model
   and link to the operator-config rotation runbook.

The PYPI_TOKEN secret is fanned out to this repo from the operator-host
SSOT by /opt/molecule-bootstrap/sync-pypi-token.sh — see operator-config
PR#48. It supersedes PR#4 (which only renamed .github -> .gitea without
fixing the OIDC issue) and unblocks the pushed v0.1.3 tag.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.gitea/workflows/publish.yml

@@ -0,0 +1,95 @@
+name: Publish to PyPI
+
+# Triggered on tag push (vX.Y.Z). Tag-on-push instead of release-creation
+# is the cheaper UX — `git tag v0.1.0 && git push origin v0.1.0` ships
+# without leaving the terminal.
+on:
+  push:
+    tags:
+      - "v[0-9]+.[0-9]+.[0-9]+"
+      - "v[0-9]+.[0-9]+.[0-9]+rc[0-9]+"
+
+# Post-2026-05-06 (Molecule-AI GitHub org suspension): PyPI's Trusted
+# Publisher OIDC flow only accepts GitHub/GitLab/Google/ActiveState
+# issuers — not Gitea. This workflow uses a long-lived PyPI API token
+# stored as the repo-level secret PYPI_TOKEN, fanned out from the
+# operator-host SSOT (/etc/molecule-bootstrap/all-credentials.env) by
+# /opt/molecule-bootstrap/sync-pypi-token.sh.
+permissions:
+  contents: read
+
+# Serialize tag-driven publishes so two concurrent tag pushes don't both
+# try to upload the same version and race PyPI.
+concurrency:
+  group: publish-pypi
+  cancel-in-progress: false
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Verify tag matches pyproject version
+        run: |
+          tag="${GITHUB_REF#refs/tags/v}"
+          pkg=$(python -c "import tomllib; print(tomllib.load(open('pyproject.toml','rb'))['project']['version'])")
+          if [ "$tag" != "$pkg" ]; then
+            echo "::error::tag $tag does not match pyproject version $pkg — aborting publish to keep PyPI in sync with git tags"
+            exit 1
+          fi
+
+      - name: Build sdist + wheel
+        run: |
+          python -m pip install --upgrade pip build twine
+          python -m build
+
+      - name: Smoke-import the built wheel
+        run: |
+          python -m venv /tmp/install-test
+          /tmp/install-test/bin/pip install dist/*.whl
+          /tmp/install-test/bin/codex-channel-molecule --help
+
+      - name: Verify package metadata (twine check)
+        run: python -m twine check dist/*
+
+      - uses: actions/upload-artifact@v3 # pinned to v3 for Gitea act_runner v0.6 compatibility (internal#46)
+        with:
+          name: dist
+          path: dist/
+
+  publish:
+    needs: build
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/download-artifact@v3 # pinned to v3 for Gitea act_runner v0.6 compatibility (internal#46)
+        with:
+          name: dist
+          path: dist/
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Publish to PyPI
+        # PYPI_TOKEN: repo-level Gitea Actions secret, written by
+        # /opt/molecule-bootstrap/sync-pypi-token.sh from the operator-host
+        # SSOT. Never set this by hand — rotate via the SSOT instead
+        # (ops/PYPI_TOKEN_ROTATION.md in operator-config).
+        env:
+          PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
+        run: |
+          if [ -z "$PYPI_TOKEN" ]; then
+            echo "::error::PYPI_TOKEN secret is not set. Run sync-pypi-token.sh on the operator host to fan it out from SSOT."
+            exit 1
+          fi
+          python -m pip install --upgrade twine
+          python -m twine upload \
+            --repository pypi \
+            --username __token__ \
+            --password "$PYPI_TOKEN" \
+            dist/*

.github/workflows/publish.yml

@@ -1,69 +0,0 @@
-name: Publish to PyPI
-
-# Triggered on tag push (vX.Y.Z). Tag-on-push instead of release-creation
-# is the cheaper UX — `git tag v0.1.0 && git push origin v0.1.0` ships
-# without leaving the terminal.
-on:
-  push:
-    tags:
-      - "v[0-9]+.[0-9]+.[0-9]+"
-      - "v[0-9]+.[0-9]+.[0-9]+rc[0-9]+"
-
-permissions:
-  contents: read
-  # OIDC token for PyPI trusted-publisher auth — no secret token needed.
-  # PyPI side: register
-  #   github.com/Molecule-AI/codex-channel-molecule
-  #   workflow=publish.yml environment=pypi
-  # under "Trusted publisher management" on the codex-channel-molecule
-  # PyPI project page (see README "Releasing" section).
-  id-token: write
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.12"
-
-      - name: Verify tag matches pyproject version
-        run: |
-          tag="${GITHUB_REF#refs/tags/v}"
-          pkg=$(python -c "import tomllib; print(tomllib.load(open('pyproject.toml','rb'))['project']['version'])")
-          if [ "$tag" != "$pkg" ]; then
-            echo "::error::tag $tag does not match pyproject version $pkg — aborting publish to keep PyPI in sync with git tags"
-            exit 1
-          fi
-
-      - name: Build sdist + wheel
-        run: |
-          python -m pip install --upgrade pip build
-          python -m build
-
-      - name: Smoke-import the built wheel
-        run: |
-          python -m venv /tmp/install-test
-          /tmp/install-test/bin/pip install dist/*.whl
-          /tmp/install-test/bin/codex-channel-molecule --help
-
-      - uses: actions/upload-artifact@v4
-        with:
-          name: dist
-          path: dist/
-
-  publish:
-    needs: build
-    runs-on: ubuntu-latest
-    environment: pypi
-    permissions:
-      id-token: write
-    steps:
-      - uses: actions/download-artifact@v4
-        with:
-          name: dist
-          path: dist/
-
-      - uses: pypa/gh-action-pypi-publish@release/v1

README.md

@@ -2,7 +2,7 @@
 
 Bridge daemon — gives [codex CLI](https://github.com/openai/codex) push parity with the [Molecule AI](https://moleculesai.com) platform's other external runtimes.
 
-The Molecule platform's [`hermes-channel-molecule`](https://github.com/Molecule-AI/hermes-channel-molecule) plugin gives `hermes-agent` true push delivery — peer agents and canvas-user messages land mid-session as conversation turns. Codex CLI has no plugin API today and its MCP runtime drops inbound notifications, so this daemon is the equivalent push surface — built outside the codex process.
+The Molecule platform's [`hermes-channel-molecule`](https://git.moleculesai.app/molecule-ai/hermes-channel-molecule) plugin gives `hermes-agent` true push delivery — peer agents and canvas-user messages land mid-session as conversation turns. Codex CLI has no plugin API today and its MCP runtime drops inbound notifications, so this daemon is the equivalent push surface — built outside the codex process.
 
 ## How it works
 
@@ -26,7 +26,7 @@ canvas user / peer agent ──► molecule platform inbox
                          canvas chat / peer workspace
 ```
 
-Per chat thread (one canvas-user thread or one peer-workspace thread) gets its own codex session_id, persisted to `~/.codex-channel-molecule/sessions.json` so daemon restarts don't lose conversation context.
+Per chat thread (one canvas-user thread or one peer-workspace thread) gets its own codex session_id, persisted to `~/.codex-channel-molecule/sessions.json` so daemon restarts don't lose conversation context. Set `CODEX_CHANNEL_MOLECULE_STATE_DIR` to override the default location (e.g. when running under systemd with a per-instance state dir).
 
 ## When to use this vs. the codex tab in the External Connect modal
 
@@ -55,6 +55,14 @@ codex-channel-molecule
 
 The daemon runs in the foreground; logs go to stderr. For systemd hosts, register a unit; for one-off use, `nohup ... &` plus a log file works.
 
+### Running under launchd / systemd (Node-on-PATH note)
+
+The codex binary is a `#!/usr/bin/env node` shim, so spawning it requires `node` to be discoverable on PATH. Under an interactive shell that's automatic; under `launchd` (macOS) and stripped-down `systemd` units PATH defaults to `/usr/bin:/bin`, and `env node` will 127-out silently.
+
+Since 0.1.2 the daemon resolves `codex` to its absolute path and prepends that directory to the subprocess PATH automatically — Node lives next to `codex` under nvm / brew / pnpm-global, so the shim's `env node` finds it. Operators typically don't need any PATH plumbing in their LaunchAgent / unit file beyond the three required env vars.
+
+If you're on an unusual install layout where `codex` and `node` live in different directories, set the LaunchAgent / unit `PATH` explicitly to include both.
+
 ## Deprecation path
 
 When [`openai/codex#17543`](https://github.com/openai/codex/issues/17543) lands upstream — a generic path for handling MCP custom notifications in codex and forwarding them into the active session as user submissions — this daemon becomes redundant. Codex itself will accept inbound molecule messages as `Op::UserInput` directly through the MCP server already wired in `~/.codex/config.toml`. Until then, this is the operator-facing answer.
@@ -62,7 +70,7 @@ When [`openai/codex#17543`](https://github.com/openai/codex/issues/17543) lands
 ## Development
 
 ```sh
-git clone https://github.com/Molecule-AI/codex-channel-molecule
+git clone https://git.moleculesai.app/molecule-ai/codex-channel-molecule.git
 cd codex-channel-molecule
 pip install -e ".[test]"
 pytest -q
@@ -72,7 +80,7 @@ Tests are entirely real-subprocess (no mocking the spawn boundary) so the boot p
 
 ## Releasing
 
-Tag-on-push triggers `publish.yml` which builds + publishes to PyPI via OIDC trusted publishing (no API token needed).
+Tag-on-push triggers `.gitea/workflows/publish.yml` which builds + publishes to PyPI via `twine upload` using the `PYPI_TOKEN` repo-level Gitea Actions secret.
 
 ```sh
 # Bump pyproject.toml `version`, commit, then:
@@ -81,19 +89,11 @@ git tag v0.1.1 && git push origin v0.1.1
 
 The workflow refuses to publish if the tag doesn't match `pyproject.toml`'s `version` — keeps PyPI versions and git tags in lockstep.
 
-**One-time PyPI setup** (before the first release):
+### Why twine, not Trusted Publisher OIDC
 
-1. Create the project on PyPI by uploading the first wheel manually, OR
-2. Pre-register the project on PyPI under a "Pending publisher" config so the first tagged push creates it.
+Post-2026-05-06 (Molecule-AI GitHub-org suspension) the canonical SCM is Gitea. PyPI's Trusted-Publisher OIDC flow only recognises GitHub / GitLab / Google / ActiveState issuers — not Gitea — so this repo (and every other PyPI-publishing repo in `molecule-ai/*`) falls back to a long-lived API token.
 
-Either way, on the project's PyPI page → "Manage" → "Publishing" → "Add a new publisher", configure:
-
-- Owner: `Molecule-AI`
-- Repository: `codex-channel-molecule`
-- Workflow filename: `publish.yml`
-- Environment name: `pypi`
-
-After this, every `git push origin v*.*.*` ships the wheel to PyPI without any further intervention.
+The `PYPI_TOKEN` secret is **not set by hand**. It is fanned out from the operator-host SSOT (`/etc/molecule-bootstrap/all-credentials.env`) by `/opt/molecule-bootstrap/sync-pypi-token.sh` (see [operator-config/etc/pypi-publishers.yaml](https://git.moleculesai.app/molecule-ai/operator-config/src/branch/main/etc/pypi-publishers.yaml)). Rotation procedure: [PYPI_TOKEN_ROTATION.md](https://git.moleculesai.app/molecule-ai/operator-config/src/branch/main/ops/PYPI_TOKEN_ROTATION.md).
 
 ## License
 

codex_channel_molecule/bridge.py

@@ -29,6 +29,14 @@ logger = logging.getLogger(__name__)
 # read timeout. 60s is the same default the MCP tool advertises.
 _LONG_POLL_SECS = 60.0
 
+# Exponential backoff bounds for wait_for_message errors. Doubles from
+# _BACKOFF_INITIAL up to _BACKOFF_MAX after each consecutive failure;
+# resets to zero on the first successful poll. Without this, a stale
+# token / DNS failure / dead platform turns into a tight 1-Hz retry
+# storm — bad citizenship across N operator daemons.
+_BACKOFF_INITIAL = 1.0
+_BACKOFF_MAX = 60.0
+
 
 class _Tools(Protocol):
     """Minimal interface the bridge needs from molecule_runtime.
@@ -51,34 +59,37 @@ class _RealTools:
     """Production implementation — calls into the installed
     molecule-ai-workspace-runtime wheel.
 
-    Imported lazily so test runs don't require the wheel.
+    The wheel is imported lazily on first instantiation so unit tests
+    that supply a fake `_Tools` don't require it.
     """
 
-    async def wait_for_message(self, timeout_secs: float) -> str:
-        from molecule_runtime.a2a_tools import tool_wait_for_message
+    def __init__(self) -> None:
+        import molecule_runtime.inbox as inbox
+        from molecule_runtime.a2a_client import PLATFORM_URL, WORKSPACE_ID
+        from molecule_runtime import a2a_tools
 
-        return await tool_wait_for_message(timeout_secs=timeout_secs)
+        state = inbox.InboxState(cursor_path=inbox.default_cursor_path())
+        inbox.activate(state)
+        inbox.start_poller_thread(state, PLATFORM_URL, WORKSPACE_ID)
+        self._tools = a2a_tools
+
+    async def wait_for_message(self, timeout_secs: float) -> str:
+        return await self._tools.tool_wait_for_message(timeout_secs=timeout_secs)
 
     async def inbox_pop(self, activity_id: str) -> str:
-        from molecule_runtime.a2a_tools import tool_inbox_pop
-
-        return await tool_inbox_pop(activity_id=activity_id)
+        return await self._tools.tool_inbox_pop(activity_id=activity_id)
 
     async def send_message_to_user(
         self, message: str, workspace_id: Optional[str]
     ) -> str:
-        from molecule_runtime.a2a_tools import tool_send_message_to_user
-
-        return await tool_send_message_to_user(
+        return await self._tools.tool_send_message_to_user(
             message=message, workspace_id=workspace_id
         )
 
     async def delegate_task(
         self, workspace_id: str, task: str, source_workspace_id: Optional[str]
     ) -> str:
-        from molecule_runtime.a2a_tools import tool_delegate_task
-
-        return await tool_delegate_task(
+        return await self._tools.tool_delegate_task(
             workspace_id=workspace_id,
             task=task,
             source_workspace_id=source_workspace_id,
@@ -207,14 +218,19 @@ async def run_bridge(
     session_store = session_store or _SessionStore(_state_dir() / "sessions.json")
 
     cycle = 0
+    backoff = 0.0
     while iterations is None or cycle < iterations:
         cycle += 1
         try:
             payload = await tools.wait_for_message(timeout_secs=_LONG_POLL_SECS)
         except Exception as exc:
-            logger.warning("wait_for_message raised: %s", exc)
-            await asyncio.sleep(1.0)
+            backoff = min(backoff * 2, _BACKOFF_MAX) if backoff else _BACKOFF_INITIAL
+            logger.warning(
+                "wait_for_message raised: %s — backing off %.1fs", exc, backoff,
+            )
+            await asyncio.sleep(backoff)
             continue
+        backoff = 0.0
 
         message = _extract_inbox_message(payload)
         if message is None:
@@ -242,6 +258,16 @@ async def _handle_one(
     )
 
     result: CodexResult = await runner.run(message=body, session_id=session_id)
+    logger.info(
+        "codex result for %s: exit=%s session=%s stdout=%d stderr_tail=%d",
+        activity_id,
+        result.exit_code,
+        result.session_id or "none",
+        len(result.text),
+        len(result.stderr_tail),
+    )
+    if result.exit_code != 0:
+        logger.warning("codex stderr tail for %s: %s", activity_id, result.stderr_tail)
     if result.session_id and result.session_id != session_id:
         store.set(chat_id, result.session_id)
 

codex_channel_molecule/codex_runner.py

@@ -69,6 +69,31 @@ class CodexRunner:
             )
         self._codex_bin = resolved
         self._timeout_secs = timeout_secs
+        # The codex binary is a `#!/usr/bin/env node` shim, so spawning
+        # it requires `node` to be discoverable via the subprocess PATH.
+        # Under a shell launch this is fine (operator's interactive PATH
+        # has Node), but launchd and most systemd unit defaults strip
+        # PATH down to /usr/bin:/bin — `env node` then 127s out and the
+        # daemon silently fails on every turn.
+        #
+        # Fix: remember the directory containing the resolved codex
+        # binary. Node lives next to it under nvm / brew / pnpm
+        # global installs. Subprocesses get that dir prepended to their
+        # PATH (see _build_env). Operators don't have to thread PATH
+        # through their LaunchAgent / unit file — just $WORKSPACE_ID,
+        # $PLATFORM_URL, $MOLECULE_WORKSPACE_TOKEN per the install
+        # contract.
+        self._codex_bin_dir = os.path.dirname(resolved)
+
+    def _build_env(self) -> dict[str, str]:
+        env = os.environ.copy()
+        existing_path = env.get("PATH", "")
+        if self._codex_bin_dir not in existing_path.split(os.pathsep):
+            env["PATH"] = (
+                self._codex_bin_dir
+                + (os.pathsep + existing_path if existing_path else "")
+            )
+        return env
 
     async def run(
         self,
@@ -81,16 +106,23 @@ class CodexRunner:
         that session. When None, codex starts a fresh session and assigns
         a new id (returned in CodexResult.session_id).
         """
-        args: list[str] = [self._codex_bin, "exec", "--skip-git-repo-check"]
         if session_id:
-            args.extend(["--resume", session_id])
-        args.append(message)
+            args = [
+                self._codex_bin,
+                "exec",
+                "resume",
+                "--skip-git-repo-check",
+                session_id,
+                message,
+            ]
+        else:
+            args = [self._codex_bin, "exec", "--skip-git-repo-check", message]
 
         proc = await asyncio.create_subprocess_exec(
             *args,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
-            env=os.environ.copy(),
+            env=self._build_env(),
         )
 
         try:

pyproject.toml

@@ -4,14 +4,26 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "codex-channel-molecule"
-version = "0.1.0"
+version = "0.1.3"
 description = "Bridge daemon for codex CLI ↔ Molecule platform — long-polls the platform inbox, runs `codex exec --resume <session>` per inbound message, replies via send_message_to_user MCP tool. Counterpart to hermes-channel-molecule."
 readme = "README.md"
 requires-python = ">=3.11"
 license = { text = "Apache-2.0" }
 authors = [{ name = "Molecule AI" }]
 dependencies = [
-    "molecule-ai-workspace-runtime>=0.1.110",
+    # Floor raised from 0.1.110 → 0.1.129 to pull in the SSOT A2A response
+    # parser (a2a_response.py, introduced 0.1.129). Pre-0.1.129 the legacy
+    # inline sniffer in a2a_client.send_a2a_message treated the poll-mode
+    # ``{"status": "queued", "delivery_mode": "poll", "method": "..."}``
+    # envelope as malformed and surfaced ``[A2A_ERROR] unexpected response
+    # shape (no result, no error): ...`` on every reply attempt — that
+    # then propagated to canvas as the workspace's task label and triggered
+    # a ~3s retry storm. The 0.1.129+ runtime classifies the envelope as
+    # ``Queued`` and short-circuits to the durable /delegate-poll path
+    # (a2a_tools_delegation._delegate_sync_via_polling). See
+    # ``molecule-ai/internal#424`` and ``molecule-core#2967`` for the
+    # full incident + fix history.
+    "molecule-ai-workspace-runtime>=0.1.129",
 ]
 
 [project.optional-dependencies]
@@ -24,8 +36,8 @@ test = [
 codex-channel-molecule = "codex_channel_molecule.daemon:main"
 
 [project.urls]
-Homepage = "https://github.com/Molecule-AI/codex-channel-molecule"
-Repository = "https://github.com/Molecule-AI/codex-channel-molecule"
+Homepage = "https://git.moleculesai.app/molecule-ai/codex-channel-molecule"
+Repository = "https://git.moleculesai.app/molecule-ai/codex-channel-molecule"
 
 [tool.setuptools.packages.find]
 include = ["codex_channel_molecule*"]

tests/test_bridge.py

@@ -297,3 +297,247 @@ async def test_a2a_multipart_text_is_concatenated(tmp_path):
 
     msg, _ = runner.calls[0]
     assert msg == "first chunk\nsecond chunk"
+
+
+# ----------------------------------------------------------------------
+# Error-path coverage — branches operators actually hit in production
+# when the platform is flaky, the message is malformed, or codex returns
+# nothing useful. Each fake supplies the exact failure shape.
+# ----------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_wait_for_message_raises_triggers_exponential_backoff(tmp_path):
+    """When wait_for_message raises (DNS failure, dead platform, expired
+    token), the bridge must back off exponentially so N daemons under a
+    common outage don't melt the platform with 1 Hz retries.
+
+    Sleep is monkeypatched to a recorder so the test runs instantly while
+    still asserting the exact 1, 2, 4, 8, 16, 32, 60-cap progression.
+    """
+    class AlwaysRaisingTools(FakeTools):
+        async def wait_for_message(self, timeout_secs: float) -> str:
+            raise ConnectionError("simulated platform outage")
+
+    tools = AlwaysRaisingTools(inbox=[])
+    runner = FakeRunner(results=[])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    sleeps: List[float] = []
+
+    async def _record_sleep(secs):
+        sleeps.append(secs)
+
+    import codex_channel_molecule.bridge as bridge_mod
+
+    orig_sleep = bridge_mod.asyncio.sleep
+    bridge_mod.asyncio.sleep = _record_sleep
+    try:
+        await run_bridge(runner=runner, tools=tools, session_store=store, iterations=8)
+    finally:
+        bridge_mod.asyncio.sleep = orig_sleep
+
+    # Expected progression: 1 → 2 → 4 → 8 → 16 → 32 → 60 (cap) → 60.
+    assert sleeps == [1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 60.0, 60.0]
+    assert runner.calls == []  # never reached codex
+
+
+@pytest.mark.asyncio
+async def test_backoff_resets_on_first_successful_poll(tmp_path):
+    """A success after consecutive failures must clear the backoff so
+    the next failure starts at 1s again — not stuck at the cap."""
+    sequence: List[Any] = ["raise", "raise", {
+        "kind": "canvas_user", "activity_id": "act-1",
+        "arrival_workspace_id": "ws-x", "text": "real",
+    }, "raise"]
+
+    class StepTools(FakeTools):
+        def __init__(self):
+            super().__init__(inbox=[])
+            self._idx = 0
+
+        async def wait_for_message(self, timeout_secs: float) -> str:
+            item = sequence[self._idx]
+            self._idx += 1
+            if item == "raise":
+                raise ConnectionError("flake")
+            return json.dumps(item)
+
+    tools = StepTools()
+    runner = FakeRunner([CodexResult(
+        text="ok", session_id="sess", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    sleeps: List[float] = []
+
+    async def _record_sleep(secs):
+        sleeps.append(secs)
+
+    import codex_channel_molecule.bridge as bridge_mod
+
+    orig_sleep = bridge_mod.asyncio.sleep
+    bridge_mod.asyncio.sleep = _record_sleep
+    try:
+        await run_bridge(runner=runner, tools=tools, session_store=store, iterations=4)
+    finally:
+        bridge_mod.asyncio.sleep = orig_sleep
+
+    # 1s after first raise, 2s after second, success in between resets,
+    # then 1s again after the trailing raise.
+    assert sleeps == [1.0, 2.0, 1.0]
+
+
+@pytest.mark.asyncio
+async def test_inbox_pop_failure_logs_but_does_not_re_run_codex(tmp_path):
+    """If pop fails after a successful reply, codex must NOT run again
+    on the same message — the platform still has the row but our turn
+    is done. We accept duplicate delivery on the next poll cycle."""
+    inbox = [{
+        "kind": "canvas_user", "activity_id": "act-pop-fail",
+        "arrival_workspace_id": "ws-x", "text": "msg",
+    }]
+
+    class PopFailTools(FakeTools):
+        async def inbox_pop(self, activity_id):
+            raise RuntimeError("simulated 503")
+
+    tools = PopFailTools(inbox)
+    runner = FakeRunner([CodexResult(
+        text="reply", session_id="sess", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    # Reply was sent; codex ran exactly once even though pop blew up.
+    assert runner.calls == [("msg", None)]
+    assert tools.canvas_replies == [("reply", "ws-x")]
+
+
+@pytest.mark.asyncio
+async def test_peer_agent_without_peer_id_drops_reply_and_acks(tmp_path):
+    """A malformed peer_agent message (no peer_id — registry lookup
+    failure or platform bug) must not crash the bridge or call
+    delegate_task with an empty workspace_id.
+
+    The row IS acked: keeping an undeliverable message in the queue
+    would loop forever on every poll. Operator sees the warning in the
+    log and can investigate — at-least-once means "we won't lose a
+    message we CAN deliver", not "we'll loop on poison forever".
+    """
+    inbox = [{
+        "kind": "peer_agent", "activity_id": "act-no-peer",
+        "text": "from a peer with no id",
+    }]
+    tools = FakeTools(inbox)
+    runner = FakeRunner([CodexResult(
+        text="reply", session_id="sess", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    assert runner.calls == [("from a peer with no id", None)]
+    assert tools.peer_replies == []  # no destination → no delegate_task call
+    assert tools.canvas_replies == []
+    assert tools.popped == ["act-no-peer"]  # poison drained
+
+
+@pytest.mark.asyncio
+async def test_unknown_kind_drops_reply_and_acks(tmp_path):
+    """Future or malformed message kinds must not crash the daemon —
+    log + drop + ack so the queue isn't blocked. A future daemon that
+    learns the new kind reads it from the platform's persistent log,
+    not from the inbox."""
+    inbox = [{
+        "kind": "future_kind_we_dont_know", "activity_id": "act-?",
+        "text": "??",
+    }]
+    tools = FakeTools(inbox)
+    runner = FakeRunner([CodexResult(
+        text="r", session_id="s", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    assert tools.canvas_replies == []
+    assert tools.peer_replies == []
+    assert tools.popped == ["act-?"]  # drained, not looped
+
+
+@pytest.mark.asyncio
+async def test_empty_codex_output_falls_back_to_placeholder(tmp_path):
+    """If codex returns empty stdout (silent failure, model refused, etc.)
+    we still send SOMETHING to the user — silent inbound never gets a
+    reply and looks like the daemon ate it."""
+    inbox = [{
+        "kind": "canvas_user", "activity_id": "act-silent",
+        "arrival_workspace_id": "ws-x", "text": "hi",
+    }]
+    tools = FakeTools(inbox)
+    runner = FakeRunner([CodexResult(
+        text="", session_id="sess", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    assert len(tools.canvas_replies) == 1
+    text, _ = tools.canvas_replies[0]
+    assert text == "(codex returned empty output)"
+    assert tools.popped == ["act-silent"]
+
+
+@pytest.mark.asyncio
+async def test_canvas_user_falls_back_to_workspace_id(tmp_path):
+    """The chat-id encoder accepts ``workspace_id`` when
+    ``arrival_workspace_id`` is absent — older platform builds use the
+    short field name. Same session keying either way."""
+    inbox = [{
+        "kind": "canvas_user", "activity_id": "act-old-shape",
+        "workspace_id": "ws-legacy",  # no arrival_workspace_id
+        "text": "msg",
+    }]
+    tools = FakeTools(inbox)
+    runner = FakeRunner([CodexResult(
+        text="ack", session_id="sess-legacy", exit_code=0, stderr_tail="",
+    )])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    assert tools.canvas_replies == [("ack", "ws-legacy")]
+    assert store.get("canvas:ws-legacy") == "sess-legacy"
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        "not json at all",                  # JSONDecodeError
+        '"a string, not an object"',         # decoded but not a dict
+        '{"timeout": true}',                 # real timeout sentinel
+        '{"no_activity_id": "field"}',        # dict but missing activity_id
+    ],
+)
+@pytest.mark.asyncio
+async def test_malformed_payloads_are_silently_dropped(tmp_path, payload):
+    """Each shape that wait_for_message can return must skip the message
+    handler without raising. Invariant comes from _extract_inbox_message."""
+    class StaticTools(FakeTools):
+        def __init__(self, payload):
+            super().__init__(inbox=[])
+            self._payload = payload
+
+        async def wait_for_message(self, timeout_secs):
+            return self._payload
+
+    tools = StaticTools(payload)
+    runner = FakeRunner(results=[])
+    store = _SessionStore(tmp_path / "sessions.json")
+
+    await run_bridge(runner=runner, tools=tools, session_store=store, iterations=2)
+
+    assert runner.calls == []
+    assert tools.popped == []

tests/test_codex_runner.py

@@ -25,7 +25,8 @@ _FAKE_CODEX_SCRIPT = textwrap.dedent("""\
     \"\"\"Fake codex CLI for tests.
 
     Behaviors keyed on argv shape and env:
-      argv: codex exec [--skip-git-repo-check] [--resume <sid>] <message>
+      argv: codex exec [--skip-git-repo-check] <message>
+      argv: codex exec resume [--skip-git-repo-check] <sid> <message>
 
     Echoes a banner to stderr (\"session: <uuid>\") and the input message
     to stdout. Honors FAKE_EXIT_CODE for failure-path tests.
@@ -35,17 +36,27 @@ _FAKE_CODEX_SCRIPT = textwrap.dedent("""\
     args = sys.argv[1:]
     assert args[0] == \"exec\", f\"unexpected first arg: {args[0]!r}\"
 
+    if \"--resume\" in args:
+        raise SystemExit(\"--resume is not supported\")
+
     resume_id = None
-    i = 1
-    while i < len(args):
-        if args[i] == \"--resume\":
-            resume_id = args[i + 1]
-            i += 2
-        elif args[i].startswith(\"--\"):
-            i += 1  # skip unrecognized flag
-        else:
-            break
-    msg = args[i] if i < len(args) else \"\"
+    if len(args) > 1 and args[1] == \"resume\":
+        i = 2
+        while i < len(args):
+            if args[i].startswith(\"--\"):
+                i += 1
+            else:
+                break
+        resume_id = args[i]
+        msg = args[i + 1] if i + 1 < len(args) else \"\"
+    else:
+        i = 1
+        while i < len(args):
+            if args[i].startswith(\"--\"):
+                i += 1  # skip unrecognized flag
+            else:
+                break
+        msg = args[i] if i < len(args) else \"\"
 
     sid = resume_id or os.environ.get(\"FAKE_NEW_SESSION_ID\", \"a1b2c3d4-1111-2222-3333-444455556666\")
     sys.stderr.write(f\"session: {sid}\\n\")
@@ -89,6 +100,8 @@ async def test_run_resumes_existing_session(fake_codex):
     result = await runner.run(message="follow up", session_id=given)
     # Resume → no new session id is captured (input is echoed back as-is).
     assert result.session_id == given
+    assert result.exit_code == 0
+    assert result.text == "echo: follow up"
 
 
 @pytest.mark.asyncio
@@ -134,5 +147,103 @@ def test_extract_session_id_matches_alternate_banner_shape():
     ) == "12345678-aaaa-bbbb-cccc-dddddddddddd"
 
 
+@pytest.mark.asyncio
+async def test_run_uses_exec_resume_subcommand_not_legacy_flag(fake_codex):
+    """Codex CLI 0.6+ rejects ``codex exec --resume <sid>``; the correct
+    invocation is ``codex exec resume [--skip-git-repo-check] <sid> <prompt>``.
+    The fake-codex fixture raises SystemExit if it sees ``--resume`` in
+    argv, so the assertion is that the call SUCCEEDS — proving the
+    runner moved off the legacy flag.
+    """
+    runner = CodexRunner(codex_bin=str(fake_codex), timeout_secs=10.0)
+    sid = "deadbeef-1111-2222-3333-444455556666"
+    result = await runner.run(message="follow", session_id=sid)
+    # Subprocess returned 0 because the fake didn't see --resume. If the
+    # runner ever regresses to the old shape, the fake exits non-zero
+    # with "--resume is not supported" → result.exit_code != 0.
+    assert result.exit_code == 0, (
+        f"Expected exec resume subcommand shape, got "
+        f"exit_code={result.exit_code} stderr_tail={result.stderr_tail!r}"
+    )
+    assert result.session_id == sid
+
+
+@pytest.mark.asyncio
+async def test_run_prepends_codex_bin_dir_to_subprocess_path(tmp_path):
+    """The codex binary is a `#!/usr/bin/env node` shim. Under launchd /
+    systemd, the parent process's PATH is stripped to /usr/bin:/bin, so
+    the shebang's `env node` lookup fails and codex 127s out silently.
+    The runner mitigates this by prepending the codex binary's own
+    directory (where node lives in nvm/brew/pnpm-global layouts) to the
+    subprocess PATH. Pin: subprocess sees that dir first in PATH.
+    """
+    bin_dir = tmp_path / "fakebin"
+    bin_dir.mkdir()
+    fake = bin_dir / "codex"
+    # Echo the resolved subprocess PATH so the test can inspect it.
+    fake.write_text(textwrap.dedent("""\
+        #!/usr/bin/env python3
+        import os, sys
+        sys.stderr.write(f"session: a1b2c3d4-1111-2222-3333-444455556666\\n")
+        sys.stdout.write(os.environ.get("PATH", ""))
+    """))
+    fake.chmod(0o755)
+
+    runner = CodexRunner(codex_bin=str(fake), timeout_secs=10.0)
+    # Strip PATH down to the launchd-default minimum to prove the
+    # runner's own injection puts bin_dir back.
+    import os as _os
+    saved = _os.environ.get("PATH", "")
+    _os.environ["PATH"] = "/usr/bin:/bin"
+    try:
+        result = await runner.run(message="probe")
+    finally:
+        _os.environ["PATH"] = saved
+
+    assert result.exit_code == 0
+    seen_path = result.text
+    # bin_dir must appear FIRST so `env node` finds it before falling
+    # back to the (empty for launchd) system PATH.
+    assert seen_path.startswith(str(bin_dir) + os.pathsep), (
+        f"expected codex_bin_dir prefix in subprocess PATH, got: {seen_path!r}"
+    )
+    assert "/usr/bin" in seen_path, (
+        f"expected existing PATH preserved after prefix, got: {seen_path!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_run_does_not_duplicate_codex_bin_dir_in_path(tmp_path):
+    """If the operator's PATH already includes the codex bin dir (the
+    shell-launched case), don't double it — keeps logs and downstream
+    tools sane.
+    """
+    bin_dir = tmp_path / "alreadythere"
+    bin_dir.mkdir()
+    fake = bin_dir / "codex"
+    fake.write_text(textwrap.dedent("""\
+        #!/usr/bin/env python3
+        import os, sys
+        sys.stderr.write(f"session: a1b2c3d4-1111-2222-3333-444455556666\\n")
+        sys.stdout.write(os.environ.get("PATH", ""))
+    """))
+    fake.chmod(0o755)
+
+    runner = CodexRunner(codex_bin=str(fake), timeout_secs=10.0)
+    import os as _os
+    saved = _os.environ.get("PATH", "")
+    _os.environ["PATH"] = f"{bin_dir}:/usr/bin:/bin"
+    try:
+        result = await runner.run(message="probe")
+    finally:
+        _os.environ["PATH"] = saved
+
+    seen_path = result.text
+    occurrences = seen_path.split(os.pathsep).count(str(bin_dir))
+    assert occurrences == 1, (
+        f"expected codex_bin_dir to appear exactly once, got {occurrences}: {seen_path!r}"
+    )
+
+
 def test_extract_session_id_returns_none_on_no_match():
     assert _extract_session_id("nothing relevant here") is None

tests/test_runtime_dependency_floor.py

@@ -0,0 +1,106 @@
+"""Pin the minimum molecule-ai-workspace-runtime version that ships
+the SSOT A2A response parser.
+
+Background — see ``molecule-ai/internal#424``:
+
+Pre-runtime-0.1.129 the inline sniffer in ``a2a_client.send_a2a_message``
+checked for ``result`` or ``error`` keys and routed everything else to
+``[A2A_ERROR] unexpected response shape (no result, no error): ...``.
+That branch fired on every reply attempt to a poll-mode peer because the
+platform proxy synthesizes the success envelope as
+``{"status": "queued", "delivery_mode": "poll", "method": "..."}``
+which has NEITHER ``result`` NOR ``error``. The retry loop hammered the
+caller's canvas with the error string every ~3s.
+
+Version 0.1.129 introduced ``a2a_response.py`` — a typed parser with an
+explicit ``Queued`` variant — and the matching ``[A2A_QUEUED]`` sentinel
+in ``a2a_client.py``. ``a2a_tools_delegation.tool_delegate_task`` then
+falls back to the durable ``/delegate`` + ``/delegations`` polling path,
+which IS the correct synchronous facade for poll-mode peers.
+
+This test asserts the floor is held at ``>=0.1.129`` so a future
+dependency-housekeeping pass cannot silently lower it back into the
+broken range. If the floor is ever raised further (e.g. to require a
+later SSOT parser feature), update the constant below — the lower bound
+must never go below 0.1.129.
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+import pytest
+
+if sys.version_info >= (3, 11):
+    import tomllib
+else:  # pragma: no cover - Python 3.11+ required by pyproject anyway
+    import tomli as tomllib
+
+
+# The earliest runtime version that ships the SSOT A2A response parser
+# with the typed ``Queued`` variant. Bumping this floor MUST be paired
+# with a comment update in pyproject.toml's dependencies block.
+_MINIMUM_RUNTIME_VERSION = (0, 1, 129)
+
+
+def _parse_version(spec: str) -> tuple[int, int, int]:
+    """Extract the lower-bound version from a dependency specifier.
+
+    Supports the common shapes used in our pyproject files:
+
+      * ``pkg>=1.2.3``         → (1, 2, 3)
+      * ``pkg>=1.2.3,<2``      → (1, 2, 3)
+      * ``pkg~=1.2.3``         → (1, 2, 3)
+
+    Raises ``ValueError`` on anything else so the test fails loud rather
+    than silently passing on a malformed specifier.
+    """
+    m = re.search(r"(?:>=|~=)\s*(\d+)\.(\d+)\.(\d+)", spec)
+    if not m:
+        raise ValueError(f"no lower-bound version in specifier: {spec!r}")
+    return (int(m.group(1)), int(m.group(2)), int(m.group(3)))
+
+
+@pytest.fixture(scope="module")
+def runtime_specifier() -> str:
+    """Return the raw dependency specifier for molecule-ai-workspace-runtime."""
+    root = Path(__file__).resolve().parent.parent
+    pyproject = root / "pyproject.toml"
+    with pyproject.open("rb") as f:
+        data = tomllib.load(f)
+    deps = data["project"]["dependencies"]
+    for dep in deps:
+        # match the package name allowing a hyphen-or-underscore in case
+        # the spec ever normalizes — PEP 503 treats them equivalently.
+        if re.match(r"^molecule[-_]ai[-_]workspace[-_]runtime\b", dep):
+            return dep
+    raise AssertionError(
+        "pyproject.toml is missing a molecule-ai-workspace-runtime dependency"
+    )
+
+
+def test_runtime_floor_includes_a2a_response_parser(runtime_specifier: str) -> None:
+    """The runtime floor must be at or above the SSOT parser release."""
+    bound = _parse_version(runtime_specifier)
+    assert bound >= _MINIMUM_RUNTIME_VERSION, (
+        f"runtime floor {bound} is below {_MINIMUM_RUNTIME_VERSION} — "
+        f"pre-0.1.129 the A2A response parser misclassifies the poll-mode "
+        f"queued envelope as malformed and surfaces "
+        f"'[A2A_ERROR] unexpected response shape' on every poll-mode peer "
+        f"reply. See molecule-ai/internal#424."
+    )
+
+
+def test_runtime_specifier_uses_a_lower_bound(runtime_specifier: str) -> None:
+    """A bare ``pkg`` or upper-only spec would silently install ANY version
+    on a fresh ``pip install`` — including the buggy pre-0.1.129 range.
+
+    Require an explicit lower bound (``>=`` or ``~=``).
+    """
+    assert re.search(r">=|~=", runtime_specifier), (
+        f"runtime dependency {runtime_specifier!r} has no lower bound — "
+        f"a fresh install could resolve to a pre-0.1.129 wheel with the "
+        f"broken poll-mode parser"
+    )