Add exact-turn DGX message steering and receipt recovery; Mac scope pending

docs/engineering/codex_dgx_worker.md

@@ -422,6 +422,37 @@ instruction does not authorise deployment of newly requested work.
 Several follow-ups queued before a coding run retain all their instructions
 and source-message identities, rather than replacing one another.
 
+The optional `codex_transport: "app-server"` replaces the coding task's `exec`
+child with one controller-owned stdio app-server under the same inherited lock.
+The default remains `exec`. The owner records the exact task/attempt/thread/turn
+in `active_turn`, preserves the configured model, effort and permission profile,
+and checks for explicitly bound steering during existing activity checkpoints.
+Queue messages stay with ordinary inbox processing. Steering is reserved before
+dispatch; an acknowledgement is recorded as accepted, never as model consumption.
+Lost acknowledgements remain uncertain and are not retried or converted to Queue.
+
+The existing owner publishes a short-lived operational observation of that exact
+binding and its task delegator. The message API resolves it only for the
+authenticated delegator in the shared organisation and rechecks the canonical
+assignment and current execution attempt before accepting Steer. A browser cannot
+publish a target. The owner refreshes its observation at existing checkpoints;
+withdrawal removes only its own binding. An unavailable owner leaves the draft
+intact and offers Queue explicitly.
+
+Admission stores the intended binding with the original message. Retries retain
+that binding and delivery key, including an ambiguous HTTP result. The owner
+records accepted, not-applied or uncertain receipts on the canonical message;
+retained receipt recovery never replays guidance. The message pane distinguishes
+stored/pending guidance from controller acceptance and does not claim model
+consumption from an acknowledgement.
+
+Production activation and real consumption evidence are separate from these
+code paths. The default `exec` owner and the current Mac stdio-only owner publish
+no target, so those routes continue to offer Queue. A Mac owner adapter/runtime
+choice remains outstanding in [PR #692](https://github.com/Strong-AI-Lab/Von/pull/692).
+Local protocol and browser fixtures do not establish provider identity,
+production activation or live consumption.
+
 The reply prompt permits relevant read-only host/repository inspection. Missing
 supplied facts do not establish that inspection is unavailable. Conditional
 coding requests are interpreted after that inspection; the inbox has no

scripts/codex_von_app_server.py

@@ -0,0 +1,290 @@
+"""Single controller-owned Codex transport; never attaches to another process.
+
+The controller supplies already authorised, durably reserved steering inputs.
+Acknowledgement means accepted input, not observed model consumption. A lost
+acknowledgement is uncertain and is never automatically retried or queued.
+This transport is opt-in; it does not enable public message admission.
+"""
+
+from __future__ import annotations
+
+import json
+import queue
+import subprocess
+import threading
+import time
+from collections import deque
+
+
+class ProtocolError(RuntimeError):
+    pass
+
+
+def run_turn(
+    *,
+    command,
+    cwd,
+    model,
+    effort,
+    prompt,
+    schema,
+    environment,
+    lock_fd,
+    events,
+    errors,
+    permission_profile=None,
+    checkpoint=lambda: None,
+    on_started=lambda: None,
+    on_active=lambda thread, turn: None,
+    steering=lambda: (),
+    on_delivery=lambda message, status: None,
+    interval=15,
+):
+    """Run one owned turn, retaining the inherited worker lock until exit.
+
+    ``steering`` yields {message_id, thread_id, turn_id, input}. It must reserve
+    each source before yielding, and never yield a reserved source again after a
+    restart. Thread/turn checks here complement the owner's task/actor checks.
+    There is no whole-turn deadline: interval only schedules observation.
+    """
+    if interval <= 0:
+        raise ValueError("Observation interval must be positive")
+    if "sol" in model.lower():
+        raise ValueError("Sol-family models are disabled")
+    args = [
+        command,
+        "app-server",
+        "--strict-config",
+        "--listen",
+        "stdio://",
+        "-c",
+        "mcp_servers.von.enabled=false",
+        "-c",
+        'approval_policy="never"',
+        "-c",
+        "model_reasoning_effort=" + json.dumps(effort),
+    ]
+    if permission_profile:
+        args += ["-c", "default_permissions=" + json.dumps(permission_profile)]
+    else:
+        args += ["-c", 'sandbox_mode="workspace-write"']
+    incoming = queue.Queue()
+    pending = {}
+    deferred = deque()
+    sequence = 0
+    thread_id = turn_id = None
+    final_text = None
+    terminal = None
+    next_observation = time.monotonic() + interval
+
+    with subprocess.Popen(
+        args,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=errors,
+        text=True,
+        env=environment,
+        cwd=cwd,
+        pass_fds=(lock_fd,),
+    ) as process:
+
+        def read_output():
+            try:
+                for line in process.stdout:
+                    incoming.put(line)
+            finally:
+                incoming.put(None)
+
+        reader = threading.Thread(target=read_output, daemon=True)
+        reader.start()
+
+        def write(value):
+            process.stdin.write(json.dumps(value) + "\n")
+            process.stdin.flush()
+
+        def request(method, params, message=None):
+            nonlocal sequence
+            sequence += 1
+            # Record before writing; even a partial write has an uncertain effect.
+            pending[sequence] = (method, message)
+            write({"id": sequence, "method": method, "params": params})
+
+        def record(value):
+            events.write(json.dumps(value) + "\n")
+            events.flush()
+
+        try:
+            on_started()
+            request(
+                "initialize",
+                {
+                    "clientInfo": {"name": "von_dgx_worker", "version": "1"},
+                    "capabilities": {"experimentalApi": True},
+                },
+            )
+            while terminal is None:
+                # Schedule from time, not empty reads: continuous tool output
+                # must not starve input delivery or archive checkpoints.
+                if time.monotonic() >= next_observation and not (turn_id and deferred):
+                    checkpoint()
+                    if turn_id:
+                        for message in steering():
+                            if (message["thread_id"], message["turn_id"]) != (
+                                thread_id,
+                                turn_id,
+                            ):
+                                on_delivery(message, "not_applied")
+                                continue
+                            request(
+                                "turn/steer",
+                                {
+                                    "threadId": thread_id,
+                                    "expectedTurnId": turn_id,
+                                    "input": message["input"],
+                                },
+                                message,
+                            )
+                    next_observation = time.monotonic() + interval
+                try:
+                    line = (
+                        deferred.popleft()
+                        if turn_id and deferred
+                        else incoming.get(
+                            timeout=max(0.001, next_observation - time.monotonic())
+                        )
+                    )
+                except queue.Empty:
+                    continue
+                if line is None:
+                    raise ProtocolError(
+                        "Owned app-server exited before turn completion"
+                    )
+                value = json.loads(line)
+                if not isinstance(value, dict):
+                    raise ProtocolError("Invalid app-server envelope")
+                if "id" in value and "method" not in value:
+                    operation = pending.pop(value["id"], None)
+                    if operation is None:
+                        raise ProtocolError("Unrecognised app-server response")
+                    method, message = operation
+                    result = value.get("result", {})
+                    if method == "turn/steer":
+                        status = "uncertain"
+                        if value.get("error", {}).get("code") in {-32600, -32601, -32602}:
+                            status = "not_applied"
+                        elif "error" not in value and result.get("turnId") == turn_id:
+                            status = "accepted"
+                        record(
+                            {
+                                "type": "steering.delivery",
+                                "message_id": message["message_id"],
+                                "thread_id": thread_id,
+                                "turn_id": turn_id,
+                                "status": status,
+                            }
+                        )
+                        on_delivery(message, status)
+                        continue
+                    if "error" in value:
+                        raise ProtocolError(f"App-server rejected {method}")
+                    if method == "initialize":
+                        write({"method": "initialized"})
+                        params = {
+                            "cwd": str(cwd),
+                            "model": model,
+                            "approvalPolicy": "never",
+                        }
+                        if not permission_profile:
+                            params["sandbox"] = "workspace-write"
+                        request("thread/start", params)
+                    elif method == "thread/start":
+                        thread_id = result["thread"]["id"]
+                        record({"type": "thread.started", "thread_id": thread_id})
+                        request(
+                            "turn/start",
+                            {
+                                "threadId": thread_id,
+                                "input": [{"type": "text", "text": prompt}],
+                                "model": model,
+                                "effort": effort,
+                                "outputSchema": schema,
+                            },
+                        )
+                    elif method == "turn/start":
+                        turn_id = result["turn"]["id"]
+                        record({"type": "turn.started", "turn_id": turn_id})
+                        on_active(thread_id, turn_id)
+                    continue
+                if "id" in value:
+                    # No approval/tool authority is granted by an RPC request.
+                    write(
+                        {
+                            "id": value["id"],
+                            "error": {
+                                "code": -32601,
+                                "message": "Interactive server requests are unsupported by this controller",
+                            },
+                        }
+                    )
+                    continue
+                method = value.get("method")
+                params = value.get("params", {})
+                if (
+                    turn_id is None
+                    and params.get("threadId") == thread_id
+                    and method in {"item/completed", "turn/completed"}
+                ):
+                    deferred.append(line)
+                    continue
+                # Retain raw notifications alongside the archive-compatible rows.
+                record({"type": "app_server.notification", "notification": value})
+                if params.get("threadId") != thread_id:
+                    continue
+                if method == "item/completed" and params.get("turnId") == turn_id:
+                    item = params.get("item", {})
+                    if (
+                        item.get("type") == "agentMessage"
+                        and item.get("phase") != "commentary"
+                    ):
+                        final_text = item.get("text")
+                elif method == "turn/completed":
+                    turn = params.get("turn", {})
+                    if turn.get("id") == turn_id:
+                        terminal = turn.get("status", "failed")
+                        record(
+                            {
+                                "type": (
+                                    "turn.completed"
+                                    if terminal == "completed"
+                                    else "turn.failed"
+                                ),
+                                "turn_id": turn_id,
+                                "status": terminal,
+                            }
+                        )
+            if terminal != "completed" or not final_text:
+                raise ProtocolError(
+                    f"Owned turn ended without a final result: {terminal}"
+                )
+            return final_text
+        finally:
+            # On controlled shutdown, close the owned process even when a local
+            # receipt/checkpoint fails rather than leave unobserved work running.
+            try:
+                process.stdin.close()
+            except OSError:
+                pass
+            if process.poll() is None:
+                process.terminate()
+                try:
+                    process.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    # Only the already-terminated child is killed; this is a
+                    # shutdown bound, never a deadline for authorised coding.
+                    process.kill()
+                    process.wait()
+            reader.join(timeout=1)
+            on_active(None, None)
+            for method, message in pending.values():
+                if method == "turn/steer":
+                    on_delivery(message, "uncertain")

scripts/codex_von_inbox.py

@@ -352,6 +352,26 @@ def prepare_attachment_inputs(config, context, run):
 
 
 def launch(config, state, lock_fd):
+    # A retained/externally-created steering message cannot be interpreted as a
+    # fresh task after its intended turn has ended. Report non-delivery without
+    # asking a model to execute it.
+    if state["context"]["message"].get("submit_mode", "queue") != "queue":
+        state["result"] = {
+            "answer": (
+                "This message was not applied to its intended active turn. "
+                "The original message and target are retained. "
+                "Choose Queue explicitly to request separate work."
+            ),
+            "task_id": "",
+            "action": "reply",
+            "deployment_requested": False,
+            "new_task": None,
+        }
+        target = state["context"]["message"].get("submit_target")
+        if isinstance(target, dict) and target.get("agent_id") == config["agent_id"]:
+            state["steering_delivery"] = {"status": "not_applied", "binding": target}
+        state["phase"] = "reporting"
+        return
     run = Path(state["run_dir"])
     run.mkdir(parents=True, exist_ok=True)
     settings = resolve_execution_settings(config, {})
@@ -420,7 +440,12 @@ def authorise_source(config, api, message):
         api.messages.project_direct_message(current), config
     ):
         raise PermissionError("Inbound message no longer available to this worker")
-    if api.messages.project_direct_message(current)["content"] != message["content"]:
+    projection = api.messages.project_direct_message(current)
+    if (
+        projection["content"] != message["content"]
+        or projection.get("submit_mode", "queue") != message.get("submit_mode", "queue")
+        or projection.get("submit_target") != message.get("submit_target")
+    ):
         raise PermissionError("Inbound message changed during reply")
     allowed, _ = api.messages.authorise_direct_message_participants(
         sender_id=config["agent_id"],
@@ -494,6 +519,14 @@ def finish(config, api, state, path):
     message = state["context"]["message"]
     state["effect_stage"] = "source_authorisation"
     authorise_source(config, api, message)
+    if state.get("steering_delivery"):
+        try:
+            from .codex_von_steering import recover_delivery, reconcile_receipt
+        except ImportError:
+            from codex_von_steering import recover_delivery, reconcile_receipt
+        state["effect_stage"] = "steering_receipt"
+        recover_delivery(state)
+        reconcile_receipt(api, state, path)
     # Check again at the effect boundary, including recovered on-disk states.
     result = state["result"] = validate_result(state["result"], state["context"])
     task_id = result["task_id"]
@@ -727,6 +760,13 @@ def tick(config, api, lock_fd):
     directory.mkdir(parents=True, exist_ok=True)
     for path in sorted(directory.glob("*.json")):
         state = json.loads(path.read_text())
+        if state["phase"] == "steering_reserved":
+            try:
+                from .codex_von_steering import recover_delivery
+            except ImportError:
+                from codex_von_steering import recover_delivery
+            recover_delivery(state)
+            write_json(path, state)
         if state["phase"] == "running":
             recover(state)
             write_json(path, state)