diff --git a/AGENTS.md b/AGENTS.md index c8caa8943..5c62f9f1c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -511,6 +511,13 @@ Before claiming completion: - leave the worktree, branch, Jira, and any live represented state consistent with the reported outcome. +Completed repository work must also send the requester a concise completion +summary in Von, under an identity for the coding agent that performed the work. +Include associated Von task links when available, what was delivered, relevant +validation and deployment evidence, and any material limitation. This is +standing authorisation for those completion messages. Verify delivery through +canonical message read-back before reporting that the summary was sent. + Update current guidance when a durable invariant or routing rule changes. Put incident-specific commands, identities, paths, and outputs in dated incident records or version control history, not in this constitution. Prefer replacing diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 48be506df..ee005a2ac 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/docs/engineering/codex_dgx_worker.md @@ -23,16 +23,20 @@ An empty check makes no model request. Each task has its own Git worktree. Each execution is a fresh subscription-backed Codex session, using the retained worktree, task and previous result for continuity. -Answer a question in its task message thread or add a comment to the task. -The current Messages UI groups messages by participants: once it contains -several task threads it may omit a thread identifier on a general reply. In -that case use a task comment to keep the reply unambiguous. General inbox -messages do not create new coding assignments in this pilot. +When the operator enables the inbox, the worker also answers direct messages +from its configured delegator in its organisation, including questions after +task completion. It uses the new message, recent participant conversation and +accessible canonical task records to identify the task, even when the Messages +UI omits a thread identifier. Status questions preserve the completed state and +record the question and answer as a task comment. A coding follow-up can reopen +the identified task; an ambiguous reference prompts a clarification. General +inbox messages do not create new coding assignments in this pilot. Publishing uses an operator-configured GitHub account and follows each task's publication authority. Missing authentication leaves changes retained and the task blocked; it must not claim that local code is published. Deployment of -the public web server requires an explicit instruction in the initial task and +the public web server requires an explicit instruction in the initial task or +the current coding follow-up, and the operator-enabled deployment command described below. The chosen GitHub account and current authentication status belong in the Jira decision surface, not in the task's text. @@ -93,8 +97,8 @@ separate. The launcher should unset `OPENAI_API_KEY` and `CODEX_API_KEY`, set the dedicated `CODEX_HOME`, and execute the installed official Codex CLI. Validate an actual agent-issued shell command before enabling polling. -Install the script and adjacent `codex_von_worker_prompt.md` from the same -reviewed revision. Run them with the project's PDM-managed Python environment. +Install the worker, `codex_von_inbox.py`, and both adjacent `_prompt.md` files +from the same reviewed revision. Run them with the project's PDM-managed Python environment. A JSON config has these fields (paths are operator-selected): ```json @@ -119,6 +123,18 @@ worktree symlink. Dependency changes needing writes outside the sandbox may require operator preparation. Do not replace `pdm.lock` or synchronise a Von project with another dependency manager. +Inbox pickup is opt-in with `inbox_enabled: true` and an explicit ISO timestamp +in `inbox_since`. Select the cutoff after inspecting outstanding messages; do +not replay old setup messages unintentionally. Reply runs use a fresh read-only +Codex process under the same controller lock. They return an answer and a +semantic choice to reply or resume an identified task; the controller rechecks +authority, records the Q/A note, and delivers an idempotent reply. The process +does not execute coding or deployment work. Resumed coding starts on a later +poll and uses the current follow-up as its instructions. An old deployment +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. + For publication, authenticate GitHub CLI in a dedicated `GH_CONFIG_DIR` outside the repository, and set that directory in both the controller and Codex launcher. Keep other host accounts separate. The adapter configures each new clone's Git @@ -213,6 +229,13 @@ On the pilot DGX, the wrapper is result and exit receipt; `worker-state/worktrees/` retains coding results. Keep these private: they can contain authorised conversation excerpts. +`worker-state/inbox/` retains reply context, process receipts and reporting +checkpoints. `worker-state/followups/` binds resumed task instructions to their +source message. Interrupted reporting reconciles the existing task comment and +message rather than repeating delivery. Empty inbox and task checks make no +model request. The five-minute poll skips while either kind of run holds the +controller lock. + To pause pickup, remove only the tagged worker crontab entry. To retry a failed run, inspect its retained work and set the Von task back to pending. Do not delete worktrees to force a retry. A full host crash preserves files and makes diff --git a/scripts/codex_von_inbox.py b/scripts/codex_von_inbox.py new file mode 100644 index 000000000..714130d4e --- /dev/null +++ b/scripts/codex_von_inbox.py @@ -0,0 +1,388 @@ +"""Durable direct-message replies for the single-host Codex worker.""" + +from __future__ import annotations + +import json +import os +import subprocess +from datetime import datetime +from pathlib import Path + +try: + from .codex_von_worker import authorised_task, fingerprint, write_json +except ImportError: + from codex_von_worker import authorised_task, fingerprint, write_json + +SCHEMA = { + "type": "object", + "properties": { + "answer": {"type": "string"}, + "task_id": {"type": "string"}, + "action": {"type": "string", "enum": ["reply", "resume_task"]}, + "deployment_requested": {"type": "boolean"}, + }, + "required": ["answer", "task_id", "action", "deployment_requested"], + "additionalProperties": False, +} + + +def timestamp(value): + return datetime.fromisoformat(str(value)) + + +def state_path(config, message_id): + return Path(config["state_root"]) / "inbox" / (fingerprint(message_id) + ".json") + + +def followup_path(config, task_id): + return Path(config["state_root"]) / "followups" / (fingerprint(task_id) + ".json") + + +def task_followup(config, task_id): + path = followup_path(config, task_id) + return json.loads(path.read_text()) if path.exists() else None + + +def already_answered(config, message_id): + path = state_path(config, message_id) + return path.exists() and json.loads(path.read_text()).get("phase") == "done" + + +def allowed_message(message, config): + return ( + message.get("sender_id") == config["delegator_id"] + and config["agent_id"] in message.get("recipient_ids", []) + and message.get("organisation_concept_id") == config["organisation_id"] + ) + + +def new_messages(config, api): + since = timestamp(config["inbox_since"]) + offset = 0 + found = [] + while True: + rows = api.messages.get_messages_for_user( + config["agent_id"], include_sent=False, limit=100, skip=offset + ) + if not rows: + break + older = False + for row in rows: + message = api.messages.project_direct_message(row) + if not message.get("sent_at") or timestamp(message["sent_at"]) < since: + older = True + continue + if allowed_message(message, config) and not already_answered( + config, message["message_id"] + ): + found.append(message) + if older or len(rows) < 100: + break + offset += len(rows) + return sorted( + found, + key=lambda message: (timestamp(message["sent_at"]), message["message_id"]), + ) + + +def context_for(config, api, message): + rows = api.messages.get_conversation_between_users( + config["agent_id"], config["delegator_id"], limit=30 + ) + conversation = [api.messages.project_direct_message(row) for row in rows] + conversation = [ + row + for row in conversation + if row.get("organisation_concept_id") == config["organisation_id"] + and row.get("sent_at") + and timestamp(row["sent_at"]) <= timestamp(message["sent_at"]) + ] + task_ids = {row.get("thread_id") for row in conversation} + task_ids.add(message.get("thread_id")) + tasks = [] + for task_id in sorted(task_ids - {None, ""}): + try: + task = api.task(task_id) + except api.tasks.TaskNotFoundError: + continue + if authorised_task(task, config) and api.native_writer(task): + tasks.append(task) + return {"message": message, "recent_messages": conversation, "tasks": tasks} + + +def validate_result(value, context): + if not isinstance(value, dict) or set(value) != set(SCHEMA["required"]): + raise ValueError("Missing inbox result fields") + if not isinstance(value["answer"], str) or not value["answer"].strip(): + raise ValueError("Empty inbox answer") + if value["action"] not in ("reply", "resume_task") or not isinstance( + value["deployment_requested"], bool + ): + raise ValueError("Invalid inbox action") + ids = {task["task_concept_id"] for task in context["tasks"]} + if value["task_id"] not in ids | {""}: + raise ValueError("Task was not in authorised context") + if value["action"] == "resume_task" and not value["task_id"]: + raise ValueError("Resuming requires an identified task") + if value["action"] == "reply" and value["deployment_requested"]: + raise ValueError("A reply cannot deploy") + return value + + +def recover(state): + run = Path(state["run_dir"]) + try: + if json.loads((run / "exit.json").read_text())["returncode"] != 0: + raise ValueError("Inbox process failed") + state["result"] = validate_result( + json.loads((run / "result.json").read_text()), state["context"] + ) + except (OSError, ValueError, KeyError, TypeError): + state["result"] = { + "answer": "I could not complete this reply. The response run was interrupted or returned an invalid result; please resend your question to retry.", + "task_id": "", + "action": "reply", + "deployment_requested": False, + } + state["phase"] = "reporting" + + +def launch(config, state, lock_fd): + run = Path(state["run_dir"]) + run.mkdir(parents=True, exist_ok=True) + write_json(run / "schema.json", SCHEMA) + write_json(run / "context.json", state["context"]) + prompt = Path(__file__).with_name("codex_von_inbox_prompt.md").read_text() + prompt += "\nContext for this reply:\n" + json.dumps(state["context"], default=str) + args = [ + config["codex_command"], + "exec", + "--strict-config", + "--model", + config["model"], + "--sandbox", + "read-only", + "--cd", + str(run), + "--skip-git-repo-check", + "--json", + "-c", + "mcp_servers.von.enabled=false", + "--output-schema", + str(run / "schema.json"), + "--output-last-message", + str(run / "result.json"), + "-", + ] + env = { + key: os.environ[key] + for key in ("HOME", "PATH", "LANG", "TERM") + if key in os.environ + } + with ( + (run / "events.jsonl").open("w") as events, + (run / "stderr.log").open("w") as errors, + ): + completed = subprocess.run( + args, + input=prompt, + text=True, + env=env, + stdout=events, + stderr=errors, + pass_fds=(lock_fd,), + check=False, + ) + write_json(run / "exit.json", {"returncode": completed.returncode}) + recover(state) + + +def finish(config, api, state, path): + message = state["context"]["message"] + # Re-read the actual inbound message and both participants' authority. + current = api.messages.get_message_for_user( + message["message_id"], config["agent_id"] + ) + if not current or not allowed_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"]: + raise PermissionError("Inbound message changed during reply") + allowed, _ = api.messages.authorise_direct_message_participants( + sender_id=config["agent_id"], + recipient_ids=[config["delegator_id"]], + organisation_concept_id=config["organisation_id"], + ) + if not allowed: + raise PermissionError("Message participants are no longer authorised") + result = state["result"] + task_id = result["task_id"] + content = result["answer"] + if task_id: + try: + task = api.task(task_id) + except api.tasks.TaskNotFoundError: + task = None + if not task or not authorised_task(task, config) or not api.native_writer(task): + # A changed assignment must not strand every subsequent inbox reply. + # Freeze the narrower report so recovery repeats the same intent. + state["result"] = result = { + "answer": "I can no longer act on the related task because its availability or assignment changed. I have left its state unchanged.", + "task_id": "", + "action": "reply", + "deployment_requested": False, + } + content = result["answer"] + task_id = "" + write_json(path, state) + if task_id: + if result["action"] == "resume_task": + # This exact follow-up is the new coding/deployment authority carrier. + followup = { + "message_id": message["message_id"], + "content": message["content"], + "deployment_requested": result["deployment_requested"], + } + prior = task_followup(config, task_id) + if not state.get("resume_applied"): + if ( + prior + and prior["message_id"] != message["message_id"] + and task["status"] == "in_progress" + ): + raise RuntimeError("A different follow-up is already running") + if ( + prior + and task["status"] == "pending" + and prior["message_id"] != message["message_id"] + ): + # Several messages can arrive before the next coding run. + # Keep every queued instruction with its source identity. + followup["content"] = ( + f"Earlier queued instructions ({prior['message_id']}):\n{prior['content']}" + f"\n\nAdditional instructions ({message['message_id']}):\n{message['content']}" + ) + followup["message_ids"] = [ + *prior.get("message_ids", [prior["message_id"]]), + message["message_id"], + ] + write_json(followup_path(config, task_id), followup) + if task["status"] != "pending": + api.tasks.update_task_status( + task_id, "pending", actor_concept_id=config["agent_id"] + ) + if api.task(task_id)["status"] != "pending": + raise RuntimeError("Task reopen read-back failed") + state["resume_applied"] = True + write_json(path, state) + content += "\n\nThe task is queued for the additional work." + note = f"Question from {config['delegator_id']}:\n{message['content']}\n\nCodex DGX answer:\n{content}" + comments = task_comments(api, task_id) + existing = next( + ( + row + for row in comments + if row.get("source", {}).get("message_id") == message["message_id"] + and row.get("source", {}).get("kind") == "codex_message_followup" + ), + None, + ) + if not existing: + api.tasks.add_task_comment( + task_id, + body=note, + author_concept_id=config["agent_id"], + source={ + "kind": "codex_message_followup", + "message_id": message["message_id"], + "question_author": config["delegator_id"], + "action": result["action"], + }, + ) + comments = task_comments(api, task_id) + if not any( + row.get("body") == note + and row.get("source", {}).get("message_id") == message["message_id"] + for row in comments + ): + raise RuntimeError("Question/answer task note read-back failed") + state.setdefault("report_content", content) + write_json(path, state) + receipt = api.messages.create_message_idempotently( + delivery_idempotency_key="codex-inbox:" + message["message_id"], + sender_id=config["agent_id"], + recipient_ids=[config["delegator_id"]], + organisation_concept_id=config["organisation_id"], + thread_id=task_id or message.get("thread_id"), + reply_to_id=message["message_id"], + subject="Codex DGX reply", + content=state["report_content"], + metadata={ + "attribution": "Sent by the Codex DGX coding worker", + "source_message_id": message["message_id"], + }, + ) + reply_id = receipt.message["concept_id"] + readback = api.messages.get_message_for_user(reply_id, config["agent_id"]) + if ( + not readback + or api.messages.project_direct_message(readback)["content"] + != state["report_content"] + ): + raise RuntimeError("Inbox reply canonical read-back failed") + state.update(phase="done", reply_id=reply_id) + write_json(path, state) + + +def task_comments(api, task_id): + comments = [] + while True: + page = api.tasks.list_task_comments(task_id, limit=500, offset=len(comments)) + rows = page["comments"] + comments.extend(rows) + if not rows or len(comments) >= page.get("total", len(comments)): + return comments + + +def tick(config, api, lock_fd): + if not config.get("inbox_enabled"): + return False + directory = Path(config["state_root"]) / "inbox" + directory.mkdir(parents=True, exist_ok=True) + for path in sorted(directory.glob("*.json")): + state = json.loads(path.read_text()) + if state["phase"] == "running": + recover(state) + write_json(path, state) + if state["phase"] == "reporting": + finish(config, api, state, path) + return True + pending = new_messages(config, api) + if not pending: + return False + message = pending[0] + path = state_path(config, message["message_id"]) + state = { + "phase": "running", + "context": context_for(config, api, message), + "run_dir": str(directory / fingerprint(message["message_id"])), + } + write_json(path, state) + try: + launch(config, state, lock_fd) + except OSError: + recover(state) + write_json(path, state) + finish(config, api, state, path) + print( + json.dumps( + { + "outcome": "message_answered", + "message_id": message["message_id"], + "reply_id": state["reply_id"], + } + ), + flush=True, + ) + return True diff --git a/scripts/codex_von_inbox_prompt.md b/scripts/codex_von_inbox_prompt.md new file mode 100644 index 000000000..00473e6e7 --- /dev/null +++ b/scripts/codex_von_inbox_prompt.md @@ -0,0 +1,26 @@ +You are Codex DGX answering a new message from your configured delegator. +The controller has supplied that exact message, recent same-organisation direct +messages, and canonical accessible task records. Answer the new message in that +context. Distinguish current instructions from historical or quoted material. +Do not invent execution or deployment evidence. Do not run coding work, change +files, send messages, deploy, or access credentials/databases in this response +run. Return your response to the controller for canonical delivery. + +Select task_id only from the supplied task records when the discussion identifies +one. For a status question, answer from its status and evidence, use action=reply, +and preserve its completion state. The controller records the exact question and +your answer as a task comment with source-message provenance, even after task +completion. A general question can have an empty task_id. + +When the new message requests additional coding or supplies the answer needed to +continue an existing task, use action=resume_task for that identified task. This +can reopen a completed task. Explain what you will work on; the controller will +verify the transition and append its receipt. A question about whether work was +finished does not by itself request more coding. Ask a concise clarification if +the task or requested change is materially ambiguous. For a new unrepresented +coding task, explain that a native task assigned to Codex DGX is needed. + +Set deployment_requested=true only if this new follow-up explicitly requests +deployment of the additional work. Historical deployment instructions do not +carry forward automatically. Return answer, task_id, action (reply or +resume_task), and deployment_requested as the required structured result. diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index eeeceee35..f02e39aab 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -172,9 +172,23 @@ def inputs(self, task): "title": task["title"], "description": task.get("description"), "comments": comments, - "replies": replies, + "replies": [ + message for message in replies if not self.inbox_answered(message) + ], + **self.followup_context(task_id), } + def inbox_answered(self, message): + if not self.config.get("inbox_enabled"): + return False + return inbox_module().already_answered(self.config, message["message_id"]) + + def followup_context(self, task_id): + if not self.config.get("inbox_enabled"): + return {} + followup = inbox_module().task_followup(self.config, task_id) + return {"followup": followup} if followup else {} + def conversation(self, task): from src.backend.integrations.internal_mcp.catalogue import ( _conversation_transcript_page, @@ -520,13 +534,18 @@ def apply_deployment(config, api, state, state_path, lock_fd): if not commit or state.get("deployment_final"): return task = api.task(state["task_id"]) + live_inputs = api.inputs(task) if task else {} eligible = ( result["status"] == "completed" and bool(task) and authorised_task(task, config) and task.get("status") in ACTIVE and api.native_writer(task) - and fingerprint(api.inputs(task)) == state.get("input_hash") + and fingerprint(live_inputs) == state.get("input_hash") + and ( + not live_inputs.get("followup") + or live_inputs["followup"].get("deployment_requested") is True + ) ) run_dir = Path(state["run_dir"]) receipt_path = run_dir / "deployment.json" @@ -575,9 +594,9 @@ def apply_deployment(config, api, state, state_path, lock_fd): receipt ) if success: - result[ - "summary" - ] += f" Deployed and verified the public server at {commit[:12]}." + result["summary"] += ( + f" Deployed and verified the public server at {commit[:12]}." + ) else: result["status"] = "blocked" result["summary"] += ( @@ -588,12 +607,23 @@ def apply_deployment(config, api, state, state_path, lock_fd): ) if recover_only: result["status"] = "blocked" - result[ - "summary" - ] += " The task changed; only the interrupted deployment was reconciled." + result["summary"] += ( + " The task changed; only the interrupted deployment was reconciled." + ) write_json(state_path, state) +def inbox_module(): + try: + from . import codex_von_inbox + except ImportError: + try: + import codex_von_inbox + except ImportError: + from scripts import codex_von_inbox + return codex_von_inbox + + def tick(config, api, lock_fd): states_dir = Path(config["state_root"]) / "tasks" states_dir.mkdir(parents=True, exist_ok=True) @@ -607,6 +637,8 @@ def tick(config, api, lock_fd): apply_deployment(config, api, state, path, lock_fd) api.finish(state) write_json(path, state) + if config.get("inbox_enabled") and inbox_module().tick(config, api, lock_fd): + return for task in api.pending(): if not authorised_task(task, config) or task.get("status") not in ACTIVE: continue diff --git a/scripts/codex_von_worker_prompt.md b/scripts/codex_von_worker_prompt.md index 8413b9a26..54639a38a 100644 --- a/scripts/codex_von_worker_prompt.md +++ b/scripts/codex_von_worker_prompt.md @@ -14,6 +14,14 @@ This is a fresh execution. Existing code and the prior result carry continuity. Inspect them before repeating work. Finish the authorised task and perform proportionate validation. Preserve unrelated files and existing changes. +If assignment.followup is present, its exact message is the current request +that resumed this task. The original title, description and prior results are +context for that follow-up. Implement the additional requested work without +repeating already completed work. For deployment, use only this follow-up's +explicit deployment instruction and deployment_requested value; an old deploy +request does not carry forward. Otherwise preserve the initial-task deployment +rule below. + The controller sends your result or question to Michael in Von, records the worktree and evidence on the task, and waits for a task-thread reply, task comment, or requeue when you need input. Do not send messages yourself or diff --git a/tests/backend/test_codex_von_inbox.py b/tests/backend/test_codex_von_inbox.py new file mode 100644 index 000000000..2aa033109 --- /dev/null +++ b/tests/backend/test_codex_von_inbox.py @@ -0,0 +1,298 @@ +"""Message replies retain task state, provenance and durable effect receipts.""" + +import copy +from types import SimpleNamespace + +import pytest + +from scripts import codex_von_inbox as inbox +from scripts import codex_von_worker as worker + + +@pytest.fixture +def fixture(tmp_path): + config = { + "agent_id": "#V#agent", + "delegator_id": "#V#owner", + "organisation_id": "#V#org", + "state_root": str(tmp_path), + "inbox_enabled": True, + "inbox_since": "2026-09-11T15:00:00Z", + } + message = { + "message_id": "#V#question", + "sender_id": "#V#owner", + "recipient_ids": ["#V#agent"], + "organisation_concept_id": "#V#org", + "sent_at": "2026-09-11T15:32:00Z", + "content": "Was this finished?", + "thread_id": None, + } + task = { + "task_concept_id": "#V#task", + "title": "Desktop tray", + "assignee_concept_id": "#V#agent", + "created_by_concept_id": "#V#owner", + "organisation_concept_id": "#V#org", + "status": "completed", + "evidence": "Deployed PR #617", + } + rows = [message] + comments, sent, transitions = [], {}, [] + + def create(**kwargs): + sent.setdefault( + kwargs["delivery_idempotency_key"], + {**kwargs, "concept_id": "#V#reply", "message_id": "#V#reply"}, + ) + return SimpleNamespace(message=sent[kwargs["delivery_idempotency_key"]]) + + def status(task_id, value, **kwargs): + transitions.append(value) + task["status"] = value + + api = SimpleNamespace( + task=lambda _: copy.deepcopy(task), + native_writer=lambda _: True, + tasks=SimpleNamespace( + TaskNotFoundError=KeyError, + list_task_comments=lambda *a, **k: {"comments": comments}, + add_task_comment=lambda *a, **k: comments.append(k), + update_task_status=status, + ), + messages=SimpleNamespace( + get_messages_for_user=lambda *a, **k: rows, + get_conversation_between_users=lambda *a, **k: [ + message, + {**message, "sender_id": "#V#agent", "thread_id": "#V#task"}, + ], + project_direct_message=lambda row: row, + get_message_for_user=lambda mid, _: ( + message + if mid == message["message_id"] + else next(iter(sent.values()), None) + ), + authorise_direct_message_participants=lambda **k: (True, []), + create_message_idempotently=create, + ), + ) + return config, api, message, task, comments, sent, transitions, rows + + +@pytest.mark.parametrize( + "action,deploy", [("reply", False), ("resume_task", False), ("resume_task", True)] +) +def test_reply_or_reopen_records_one_question_answer_and_one_message( + fixture, monkeypatch, action, deploy +): + config, api, message, task, comments, sent, transitions, _ = fixture + launches = [] + + def launch(config, state, fd): + launches.append(state) + state.update( + phase="reporting", + result={ + "answer": "The tray was deployed.", + "task_id": "#V#task", + "action": action, + "deployment_requested": deploy, + }, + ) + + monkeypatch.setattr(inbox, "launch", launch) + assert inbox.tick(config, api, 0) + assert not inbox.tick(config, api, 0) + assert len(launches) == len(comments) == len(sent) == 1 + assert ( + "Was this finished?" in comments[0]["body"] + and "The tray was deployed." in comments[0]["body"] + ) + assert comments[0]["source"]["message_id"] == message["message_id"] + assert task["status"] == ("pending" if action == "resume_task" else "completed") + assert transitions == (["pending"] if action == "resume_task" else []) + if action == "resume_task": + assert inbox.task_followup(config, "#V#task") == { + "message_id": "#V#question", + "content": message["content"], + "deployment_requested": deploy, + } + else: + assert inbox.task_followup(config, "#V#task") is None + + +def test_crash_after_note_and_delivery_reconciles_without_duplicate( + fixture, monkeypatch +): + config, api, message, _, comments, sent, _, _ = fixture + state = { + "phase": "reporting", + "context": inbox.context_for(config, api, message), + "result": { + "answer": "Yes, deployed.", + "task_id": "#V#task", + "action": "reply", + "deployment_requested": False, + }, + } + path = inbox.state_path(config, message["message_id"]) + original = inbox.write_json + + def crash(path, value): + if value.get("phase") == "done": + raise OSError("simulated loss after delivery") + original(path, value) + + monkeypatch.setattr(inbox, "write_json", crash) + with pytest.raises(OSError): + inbox.finish(config, api, state, path) + monkeypatch.setattr(inbox, "write_json", original) + assert inbox.tick(config, api, 0) + assert len(comments) == len(sent) == 1 + + +def test_reassigned_task_gets_honest_reply_without_blocking_inbox(fixture): + config, api, message, task, comments, sent, transitions, _ = fixture + state = { + "phase": "reporting", + "context": inbox.context_for(config, api, message), + "result": { + "answer": "I will resume it.", + "task_id": "#V#task", + "action": "resume_task", + "deployment_requested": False, + }, + } + task["assignee_concept_id"] = "#V#other" + inbox.finish(config, api, state, inbox.state_path(config, message["message_id"])) + assert state["phase"] == "done" + assert "assignment changed" in next(iter(sent.values()))["content"] + assert not comments and not transitions + assert not inbox.tick(config, api, 0) + + +def test_two_queued_followups_preserve_both_coding_requests(fixture): + config, api, message, task, comments, sent, transitions, _ = fixture + + def report(): + state = { + "phase": "reporting", + "context": inbox.context_for(config, api, message), + "result": { + "answer": "I will make this change.", + "task_id": "#V#task", + "action": "resume_task", + "deployment_requested": False, + }, + } + inbox.finish( + config, api, state, inbox.state_path(config, message["message_id"]) + ) + + message["content"] = "Move the send controls beside the input." + report() + message["message_id"] = "#V#second_question" + message["content"] = "Also use compact identity labels." + api.messages.get_message_for_user = lambda mid, _: ( + message if mid == message["message_id"] else list(sent.values())[-1] + ) + report() + followup = inbox.task_followup(config, "#V#task") + assert "Move the send controls" in followup["content"] + assert "compact identity labels" in followup["content"] + assert followup["message_ids"] == ["#V#question", "#V#second_question"] + assert task["status"] == "pending" and len(comments) == 2 + assert transitions == ['pending'] + + +def test_empty_old_or_foreign_mail_does_not_invoke_model(fixture, monkeypatch): + config, api, message, _, comments, sent, _, rows = fixture + rows[:] = [ + {**message, "sender_id": "#V#other"}, + {**message, "organisation_concept_id": "#V#other"}, + {**message, "sent_at": "2026-09-10T01:00:00Z"}, + ] + monkeypatch.setattr( + inbox, "launch", lambda *a: pytest.fail("Unexpected model call") + ) + assert not inbox.tick(config, api, 0) + assert not comments and not sent + + +def test_unknown_task_and_reply_deployment_are_denied(fixture): + config, api, message, *_ = fixture + context = inbox.context_for(config, api, message) + for result in [ + { + "answer": "Done", + "task_id": "#V#outsider", + "action": "resume_task", + "deployment_requested": False, + }, + { + "answer": "Done", + "task_id": "#V#task", + "action": "reply", + "deployment_requested": True, + }, + ]: + with pytest.raises(ValueError): + inbox.validate_result(result, context) + + +def test_resume_without_new_deployment_instruction_blocks_old_deploy_request( + fixture, tmp_path +): + config, api, _message, task, *_ = fixture + task["status"] = "in_progress" + inputs = { + "description": "Implement and deploy the original task", + "followup": {"content": "Adjust the icon", "deployment_requested": False}, + } + api.inputs = lambda _: inputs + config["deployment_command"] = "/should/not/run" + run = tmp_path / "run" + run.mkdir() + state = { + "task_id": "#V#task", + "run_dir": str(run), + "input_hash": worker.fingerprint(inputs), + "result": { + "status": "completed", + "summary": "Changed icon", + "deploy_commit": "a" * 40, + }, + } + worker.apply_deployment(config, api, state, tmp_path / "state.json", 0) + assert state["deployment_final"] == "not_authorised" + + +def test_inbox_launch_is_read_only_and_inherits_the_single_worker_lock( + fixture, monkeypatch +): + config, api, message, *_ = fixture + config.update(codex_command="codex", model="gpt-5.6-terra") + state = { + "run_dir": str(inbox.state_path(config, "#V#question").with_suffix("")), + "context": inbox.context_for(config, api, message), + } + + def run(args, **kwargs): + assert args[args.index("--sandbox") + 1] == "read-only" + assert kwargs["pass_fds"] == (17,) + assert not set(kwargs["env"]) & {"OPENAI_API_KEY", "MONGO_URI"} + output = inbox.Path(args[args.index("--output-last-message") + 1]) + inbox.write_json( + output, + { + "answer": "Yes, deployed.", + "task_id": "#V#task", + "action": "reply", + "deployment_requested": False, + }, + ) + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(inbox.subprocess, "run", run) + inbox.launch(config, state, 17) + assert state["phase"] == "reporting"