5985 lines
280 KiB
Python
5985 lines
280 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import tomllib
|
|
import unittest
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
from unittest import mock
|
|
|
|
import mmo_runtime
|
|
import mmo_state
|
|
import mmo_workspace
|
|
import worker_runner
|
|
from common import ROOT, RuntimeSandbox, create_access_lab_session, root_thread_binding
|
|
from mmo_app_server import (
|
|
ControlDeliveryUnknown,
|
|
app_server_socket_path,
|
|
normalize_turn_failure,
|
|
)
|
|
from mmo_codex_home import _snapshot_guidance_text
|
|
from mmo_diagnostics import (
|
|
_dynamic_tool_aliases,
|
|
_root_harness_prompt,
|
|
_successful_mcp_tools,
|
|
smoke_profile,
|
|
)
|
|
from mmo_gateway import gateway_models, route_telemetry
|
|
from mmo_guidance import (
|
|
PROFILE_SKILL_NAME,
|
|
PROFILE_SKILL_RELATIVE_PATH,
|
|
agent_guidance_relative_path,
|
|
)
|
|
from mmo_profiles import AGENT_MCP_CONTROL_TOOLS, clone_profile
|
|
from mmo_runtime import (
|
|
AdmissionError,
|
|
accept_result,
|
|
begin_resume_run,
|
|
cancel_job,
|
|
cancel_session,
|
|
clean_state,
|
|
compact_session,
|
|
continue_session,
|
|
control_job,
|
|
create_session,
|
|
detach_session,
|
|
finish_session,
|
|
fork_job,
|
|
inspect_job,
|
|
integrate_patch,
|
|
iter_session_runs,
|
|
iter_sessions,
|
|
launch_interactive,
|
|
list_jobs,
|
|
load_job,
|
|
load_session,
|
|
load_session_run,
|
|
mark_session_running,
|
|
pause_session,
|
|
read_agent_trace_record,
|
|
read_result,
|
|
read_trace,
|
|
reject_result,
|
|
resolve_resume_session,
|
|
resume_interactive,
|
|
run_root_exec,
|
|
session_environment,
|
|
spawn_job,
|
|
spawn_jobs,
|
|
stop_session,
|
|
taint_session,
|
|
update_session,
|
|
wait_for_jobs,
|
|
)
|
|
from mmo_snapshot import compile_profile, load_snapshot
|
|
from mmo_state import publish_job_record, read_job_record, root_mcp_token
|
|
from mmo_util import (
|
|
append_jsonl,
|
|
atomic_write_json,
|
|
file_lock,
|
|
process_alive,
|
|
process_group_alive,
|
|
process_matches,
|
|
process_start_token,
|
|
read_json,
|
|
read_toml,
|
|
sha256_file,
|
|
toml_dumps,
|
|
)
|
|
|
|
|
|
def mcp_exchange(
|
|
session: dict,
|
|
caller: str,
|
|
*,
|
|
caller_token: str | None = None,
|
|
caller_job_id: str | None = None,
|
|
native_token: str | None = None,
|
|
run_id: str | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = {
|
|
**os.environ,
|
|
"MMO_ROOT_SESSION_ID": session["session_id"],
|
|
"MMO_RUN_ID": run_id or session["current_run_id"],
|
|
"MMO_CALLER_AGENT": caller,
|
|
"MMO_CALLER_TOKEN": caller_token or root_mcp_token(session["session_id"]),
|
|
"MMO_INSTALL_ROOT": str(ROOT),
|
|
}
|
|
if native_token is not None:
|
|
env.update({"MMO_CALLER_NATIVE": "1", "MMO_NATIVE_CALLER_TOKEN": native_token})
|
|
if caller_job_id is not None:
|
|
env["MMO_CALLER_JOB_ID"] = caller_job_id
|
|
request = (
|
|
"\n".join(
|
|
[
|
|
json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2025-06-18",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "1"},
|
|
},
|
|
}
|
|
),
|
|
json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}),
|
|
json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}),
|
|
]
|
|
)
|
|
+ "\n"
|
|
)
|
|
return subprocess.run(
|
|
[str(ROOT / "libexec" / "mmo_mcp.py")],
|
|
input=request,
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
def mcp_tool_call(
|
|
session: dict, caller: str, name: str, arguments: dict
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = {
|
|
**os.environ,
|
|
"MMO_ROOT_SESSION_ID": session["session_id"],
|
|
"MMO_RUN_ID": session["current_run_id"],
|
|
"MMO_CALLER_AGENT": caller,
|
|
"MMO_CALLER_TOKEN": root_mcp_token(session["session_id"]),
|
|
"MMO_INSTALL_ROOT": str(ROOT),
|
|
}
|
|
request = "\n".join(
|
|
[
|
|
json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2025-06-18",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "1"},
|
|
},
|
|
}
|
|
),
|
|
json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}),
|
|
json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 2,
|
|
"method": "tools/call",
|
|
"params": {"name": name, "arguments": arguments},
|
|
}
|
|
),
|
|
]
|
|
)
|
|
return subprocess.run(
|
|
[str(ROOT / "libexec" / "mmo_mcp.py")],
|
|
input=request + "\n",
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
def mcp_raw_exchange(
|
|
session: dict, caller: str, messages: list[str]
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = {
|
|
**os.environ,
|
|
"MMO_ROOT_SESSION_ID": session["session_id"],
|
|
"MMO_RUN_ID": session["current_run_id"],
|
|
"MMO_CALLER_AGENT": caller,
|
|
"MMO_CALLER_TOKEN": root_mcp_token(session["session_id"]),
|
|
"MMO_INSTALL_ROOT": str(ROOT),
|
|
}
|
|
return subprocess.run(
|
|
[str(ROOT / "libexec" / "mmo_mcp.py")],
|
|
input="\n".join(messages) + "\n",
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
class RuntimeTests(unittest.TestCase):
|
|
def test_cold_pause_rejects_bootstrap_without_a_durable_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
try:
|
|
self.assertEqual(load_session(session["session_id"])["status"], "starting")
|
|
with self.assertRaisesRegex(RuntimeError, "cannot pause session from status"):
|
|
pause_session(session["session_id"])
|
|
self.assertEqual(load_session(session["session_id"])["status"], "starting")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_provider_failure_normalization_preserves_ambiguous_reset_time(self) -> None:
|
|
failure = normalize_turn_failure(
|
|
{
|
|
"id": "turn-limit",
|
|
"status": "failed",
|
|
"error": {
|
|
"message": (
|
|
"Provider usage limit reached; limit will reset at 2026-08-22 05:24:28"
|
|
)
|
|
},
|
|
}
|
|
)
|
|
self.assertIsNotNone(failure)
|
|
assert failure is not None
|
|
self.assertEqual(failure["kind"], "provider_usage_limited")
|
|
self.assertTrue(failure["retryable"])
|
|
self.assertEqual(failure["retry_at_raw"], "2026-08-22 05:24:28")
|
|
self.assertNotIn("retry_at", failure)
|
|
|
|
zoned = normalize_turn_failure(
|
|
{
|
|
"id": "turn-limit-zoned",
|
|
"status": "failed",
|
|
"error": {"message": "quota reset at 2026-08-22 05:24:28 UTC"},
|
|
}
|
|
)
|
|
assert zoned is not None
|
|
self.assertEqual(zoned["retry_at"], "2026-08-22T05:24:28+00:00")
|
|
self.assertEqual(zoned["retry_at_timezone"], "UTC")
|
|
|
|
lowercase_zoned = normalize_turn_failure(
|
|
{
|
|
"id": "turn-limit-lowercase-zoned",
|
|
"status": "failed",
|
|
"error": {"message": "quota reset at 2026-08-22 05:24:28 utc"},
|
|
}
|
|
)
|
|
assert lowercase_zoned is not None
|
|
self.assertEqual(lowercase_zoned["retry_at"], "2026-08-22T05:24:28+00:00")
|
|
self.assertEqual(lowercase_zoned["retry_at_timezone"], "utc")
|
|
|
|
malformed = normalize_turn_failure(
|
|
{
|
|
"id": "turn-tools",
|
|
"status": "failed",
|
|
"error": {
|
|
"message": "Failed to parse tool call arguments",
|
|
"codexErrorInfo": {"type": "upstream"},
|
|
},
|
|
}
|
|
)
|
|
assert malformed is not None
|
|
self.assertEqual(malformed["kind"], "malformed_tool_arguments")
|
|
|
|
transport = normalize_turn_failure(
|
|
{
|
|
"id": "turn-transport",
|
|
"status": "failed",
|
|
"error": {
|
|
"message": "upstream stream ended",
|
|
"codexErrorInfo": {"responseStreamDisconnected": {"httpStatusCode": 503}},
|
|
},
|
|
}
|
|
)
|
|
assert transport is not None
|
|
self.assertEqual(transport["kind"], "provider_transport")
|
|
self.assertTrue(transport["retryable"])
|
|
|
|
for codex_error_info in (
|
|
"badRequest",
|
|
{"activeTurnNotSteerable": {"turnKind": "review"}},
|
|
):
|
|
terminal = normalize_turn_failure(
|
|
{
|
|
"id": "turn-terminal",
|
|
"status": "failed",
|
|
"error": {
|
|
"message": "request cannot be retried unchanged",
|
|
"codexErrorInfo": codex_error_info,
|
|
},
|
|
}
|
|
)
|
|
assert terminal is not None
|
|
self.assertEqual(terminal["kind"], "turn_failed")
|
|
self.assertFalse(terminal["retryable"])
|
|
|
|
def test_provider_limit_suspends_root_and_worker_without_erasing_evidence(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
root_result = run_root_exec(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
prompt="Retain the provider failure. FAKE_PROVIDER_USAGE_LIMIT",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(root_result["status"], "suspended", root_result)
|
|
root_session = load_session(root_result["session"]["session_id"])
|
|
self.assertEqual(root_session["failure"]["kind"], "provider_usage_limited")
|
|
self.assertEqual(root_session["failure"]["retry_at_raw"], "2026-08-22 05:24:28")
|
|
self.assertTrue(Path(root_session["partial_result_path"]).is_file())
|
|
self.assertTrue(root_result["session"]["runtime_current"])
|
|
cancel_session(root_session["session_id"])
|
|
|
|
terminal_root = run_root_exec(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
prompt="Retain the terminal provider failure. FAKE_PROVIDER_BAD_REQUEST",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(terminal_root["status"], "failed", terminal_root)
|
|
terminal_session = load_session(terminal_root["session"]["session_id"])
|
|
self.assertEqual(terminal_session["failure"]["kind"], "turn_failed")
|
|
self.assertFalse(terminal_session["failure"]["retryable"])
|
|
self.assertEqual(
|
|
terminal_session["error"],
|
|
"Provider rejected the request as invalid",
|
|
)
|
|
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Retain the provider failure. FAKE_PROVIDER_USAGE_LIMIT",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(500):
|
|
worker = load_job(job["job_id"])
|
|
if worker["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("provider-limited worker did not become suspended")
|
|
self.assertEqual(worker["failure"]["kind"], "provider_usage_limited")
|
|
self.assertEqual(worker["failure"]["retry_at_raw"], "2026-08-22 05:24:28")
|
|
self.assertTrue(Path(worker["partial_result_path"]).is_file())
|
|
self.assertTrue(mmo_runtime.public_job(worker)["runtime_current"])
|
|
app_server_pid = worker.get("app_server_pid")
|
|
app_server_token = worker.get("app_server_start_token")
|
|
deadline = time.monotonic() + 5
|
|
while process_matches(app_server_pid, app_server_token):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
self.assertFalse(process_group_alive(int(worker["app_server_pgid"])))
|
|
finally:
|
|
if load_job(job["job_id"])["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_retryable_root_failure_continues_on_a_replacement_host(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
first = run_root_exec(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
prompt=(
|
|
"Recover this goal once. FAKE_PROVIDER_USAGE_LIMIT_ONCE FAKE_SLEEP_SECONDS=0.2"
|
|
),
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(first["status"], "suspended", first)
|
|
before = load_session(first["session"]["session_id"])
|
|
old_pid = int(before["root_pid"])
|
|
old_token = str(before["root_start_token"])
|
|
thread_id = str(before["root_thread_id"])
|
|
|
|
continued = continue_session(
|
|
before["session_id"],
|
|
input_text="Continue after the transient provider failure.",
|
|
)
|
|
self.assertEqual(continued["session"]["status"], "running")
|
|
deadline = time.monotonic() + 15
|
|
while load_session(before["session_id"])["status"] not in (
|
|
mmo_runtime.TERMINAL_SESSION_STATUSES
|
|
):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(before["session_id"])
|
|
self.assertEqual(completed["status"], "completed", completed)
|
|
self.assertEqual(completed["root_thread_id"], thread_id)
|
|
self.assertFalse(process_matches(old_pid, old_token))
|
|
self.assertIsNone(completed.get("failure"))
|
|
self.assertTrue(Path(completed["result_path"]).is_file())
|
|
|
|
def test_attached_provider_limit_keeps_host_until_explicit_detach(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
session_kind="interactive",
|
|
)
|
|
client_token = process_start_token(os.getpid())
|
|
self.assertIsInstance(client_token, str)
|
|
update_session(
|
|
session["session_id"],
|
|
root_initial_prompt="Retain this attached failure. FAKE_PROVIDER_USAGE_LIMIT",
|
|
root_sandbox_mode="workspace-write",
|
|
root_client_pid=os.getpid(),
|
|
root_client_start_token=client_token,
|
|
)
|
|
started = mmo_runtime._start_root_runner(load_session(session["session_id"]))
|
|
try:
|
|
deadline = time.monotonic() + 15
|
|
while True:
|
|
current = load_session(session["session_id"])
|
|
if isinstance(current.get("failure"), dict) and isinstance(
|
|
current.get("partial_result_path"), str
|
|
):
|
|
break
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
self.assertEqual(current["status"], "running")
|
|
self.assertEqual(current["failure"]["kind"], "provider_usage_limited")
|
|
self.assertTrue(Path(current["partial_result_path"]).is_file())
|
|
self.assertTrue(
|
|
process_matches(current.get("root_pid"), current.get("root_start_token"))
|
|
)
|
|
self.assertTrue(
|
|
process_matches(
|
|
current.get("root_app_server_pid"),
|
|
current.get("root_app_server_start_token"),
|
|
)
|
|
)
|
|
|
|
detached = detach_session(session["session_id"])
|
|
self.assertEqual(detached["session"]["status"], "suspended")
|
|
deadline = time.monotonic() + 10
|
|
while process_matches(started.get("root_pid"), started.get("root_start_token")):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
suspended = load_session(session["session_id"])
|
|
self.assertEqual(suspended["status"], "suspended")
|
|
self.assertEqual(suspended["failure"]["kind"], "provider_usage_limited")
|
|
finally:
|
|
current = load_session(session["session_id"])
|
|
if current["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
cancel_session(session["session_id"])
|
|
|
|
def test_interrupted_root_cold_pause_is_reconciled_before_resume(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
update_session(
|
|
session["session_id"],
|
|
status="paused",
|
|
cold_pause_pending=True,
|
|
root_pid=12345,
|
|
root_pgid=12345,
|
|
root_start_token="root-token",
|
|
root_app_server_pid=12346,
|
|
root_app_server_pgid=12346,
|
|
root_app_server_start_token="app-token",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000087"),
|
|
)
|
|
try:
|
|
with mock.patch.object(mmo_runtime, "terminate_root_host") as terminate:
|
|
recovered = load_session(session["session_id"])
|
|
terminate.assert_called_once()
|
|
self.assertEqual(recovered["status"], "paused")
|
|
self.assertNotIn("cold_pause_pending", recovered)
|
|
self.assertNotIn("root_pid", recovered)
|
|
self.assertNotIn("root_app_server_pid", recovered)
|
|
self.assertTrue(Path(recovered["partial_result_path"]).is_file())
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_state_enumeration_is_pure_until_runtime_reconciliation(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
update_session(
|
|
session["session_id"],
|
|
status="running",
|
|
root_pid=999_999_999,
|
|
root_pgid=999_999_999,
|
|
root_start_token="definitely-not-live",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000088"),
|
|
)
|
|
try:
|
|
records = mmo_state.iter_session_records()
|
|
selected = next(
|
|
item for item in records if item["session_id"] == session["session_id"]
|
|
)
|
|
self.assertEqual(selected["status"], "running")
|
|
self.assertEqual(
|
|
mmo_state.read_session_record(mmo_state.session_dir(session["session_id"]))[
|
|
"status"
|
|
],
|
|
"running",
|
|
)
|
|
|
|
reconciled = load_session(session["session_id"])
|
|
self.assertEqual(reconciled["status"], "suspended")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_compaction_start_failure_returns_session_to_cold_pause(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
update_session(
|
|
session["session_id"],
|
|
status="paused",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000086"),
|
|
)
|
|
with (
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"_begin_resume_run_locked",
|
|
side_effect=RuntimeError("synthetic resume failure"),
|
|
),
|
|
self.assertRaisesRegex(RuntimeError, "safely cold-paused"),
|
|
):
|
|
compact_session(session["session_id"])
|
|
recovered = load_session(session["session_id"])
|
|
self.assertEqual(recovered["status"], "paused")
|
|
self.assertNotIn("cold_pause_pending", recovered)
|
|
self.assertIn("synthetic resume failure", recovered["root_compaction_error"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_terminal_root_wins_a_cold_pause_retirement_race(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
|
|
def complete_during_pause(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
return {"result": {"paused": True}}
|
|
|
|
with (
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"_root_control_socket",
|
|
return_value=Path("/tmp/fake-root-control.sock"),
|
|
),
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"send_control_request",
|
|
side_effect=complete_during_pause,
|
|
),
|
|
mock.patch.object(mmo_runtime, "_force_retire_recorded_groups"),
|
|
):
|
|
paused = pause_session(session["session_id"])
|
|
|
|
self.assertEqual(paused["session"]["status"], "completed")
|
|
final = load_session(session["session_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertNotIn("cold_pause_pending", final)
|
|
|
|
def test_terminal_worker_wins_a_cold_pause_retirement_race(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Remain active for the pause race. FAKE_SLEEP_SECONDS=20",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(500):
|
|
worker = load_job(job["job_id"])
|
|
if worker.get("control_socket_ready") and isinstance(
|
|
worker.get("active_turn_id"), str
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose a controllable turn")
|
|
|
|
def complete_during_pause(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
|
directory = mmo_state.job_dir(job["job_id"])
|
|
with file_lock(mmo_state.runtime_lock_path()):
|
|
current = read_job_record(directory)
|
|
Path(current["result_path"]).write_text("terminal result\n", encoding="utf-8")
|
|
current.update(
|
|
status="completed",
|
|
finished_at=mmo_runtime.utc_now(),
|
|
result_kind="final",
|
|
result_state="unread",
|
|
contract_valid=True,
|
|
)
|
|
publish_job_record(directory, current)
|
|
return {"result": {"paused": True}}
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"send_control_request",
|
|
side_effect=complete_during_pause,
|
|
):
|
|
paused = pause_session(session["session_id"])
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertNotIn(job["job_id"], paused["affected_jobs"])
|
|
self.assertNotIn(job["job_id"], paused["session"].get("paused_job_ids", []))
|
|
self.assertNotIn("cold_pause_pending", final)
|
|
finally:
|
|
current_session = load_session(session["session_id"])
|
|
if current_session["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_previous_package_session_is_rejected_in_place(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
path = box.state / "sessions" / session["session_id"] / "session.json"
|
|
current = read_json(path)
|
|
atomic_write_json(path, {**current, "package_version": "7.0.0"})
|
|
try:
|
|
with self.assertRaisesRegex(ValueError, "persistent session package must be"):
|
|
load_session(session["session_id"])
|
|
finally:
|
|
atomic_write_json(path, current)
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_resume_last_rejects_stale_package_state(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
current = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
update_session(
|
|
current["session_id"],
|
|
status="paused",
|
|
last_active_at="2026-08-20T00:00:00+00:00",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000089"),
|
|
)
|
|
archival = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
archival_path = box.state / "sessions" / archival["session_id"] / "session.json"
|
|
archival_current = read_json(archival_path)
|
|
atomic_write_json(
|
|
archival_path,
|
|
{
|
|
**archival_current,
|
|
"package_version": "7.0.0",
|
|
"status": "paused",
|
|
"last_active_at": "2026-08-21T00:00:00+00:00",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000090"),
|
|
},
|
|
)
|
|
try:
|
|
with self.assertRaisesRegex(RuntimeError, "invalid session state"):
|
|
resolve_resume_session(last=True, cwd=box.workspace)
|
|
finally:
|
|
atomic_write_json(archival_path, archival_current)
|
|
finish_session(current["session_id"], exit_code=0)
|
|
finish_session(archival["session_id"], exit_code=0)
|
|
|
|
def test_released_root_is_readmitted_before_resume(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
update_session(
|
|
session["session_id"],
|
|
status="paused",
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000091"),
|
|
)
|
|
try:
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"_assert_resource_capacity",
|
|
side_effect=RuntimeError("synthetic root capacity exhausted"),
|
|
) as capacity:
|
|
with self.assertRaisesRegex(RuntimeError, "root capacity exhausted"):
|
|
begin_resume_run(session["session_id"])
|
|
self.assertEqual(capacity.call_count, 1)
|
|
self.assertEqual(load_session(session["session_id"])["status"], "paused")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_live_logical_pauses_retain_capacity_until_hosts_retire(self) -> None:
|
|
session = {
|
|
"status": "paused",
|
|
"root_resource_lock_key": "shared-route",
|
|
"root_resource_units": 2,
|
|
"root_pid": 101,
|
|
"root_start_token": "root-token",
|
|
}
|
|
job = {
|
|
"status": "paused",
|
|
"resource_lock_key": "shared-route",
|
|
"resource_units": 3,
|
|
"runner_pid": 202,
|
|
"runner_start_token": "worker-token",
|
|
}
|
|
with mock.patch.object(mmo_runtime, "process_matches", return_value=True):
|
|
self.assertEqual(
|
|
mmo_runtime._active_resource_usage(sessions=[session], jobs=[job]),
|
|
{"shared-route": 5},
|
|
)
|
|
with mock.patch.object(mmo_runtime, "process_matches", return_value=False):
|
|
self.assertEqual(
|
|
mmo_runtime._active_resource_usage(sessions=[session], jobs=[job]),
|
|
{},
|
|
)
|
|
|
|
def test_cold_pause_compact_and_continue_preserve_exact_threads_and_workers(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "cold-pause-profile"
|
|
shutil.copytree(ROOT / "profiles" / "incident-hypothesis-triage", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "cold-pause-profile"
|
|
root_agent = profile_data["agents"]["incident_lead"]
|
|
root_agent["execution_mode"] = "turn"
|
|
root_agent.pop("goal_token_budget")
|
|
root_agent.pop("max_goal_token_budget")
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
root_results: list[dict[str, Any]] = []
|
|
root_failures: list[BaseException] = []
|
|
|
|
def run_root() -> None:
|
|
try:
|
|
root_results.append(
|
|
run_root_exec(
|
|
profile=profile,
|
|
cwd=box.workspace,
|
|
prompt="Complete after a cold pause. FAKE_SLEEP_SECONDS=4",
|
|
wall_timeout_seconds=30,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
root_failures.append(exc)
|
|
|
|
root_thread = threading.Thread(target=run_root)
|
|
root_thread.start()
|
|
session_id: str | None = None
|
|
worker_id: str | None = None
|
|
try:
|
|
for _ in range(500):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and isinstance(current.get("active_root_turn_id"), str)
|
|
and current.get("root_control_socket_ready")
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("root did not expose a controllable turn")
|
|
session_id = str(current["session_id"])
|
|
root_thread_id = str(current["root_thread_id"])
|
|
old_root_pid = int(current["root_pid"])
|
|
old_root_token = str(current["root_start_token"])
|
|
|
|
job = spawn_job(
|
|
session_id=session_id,
|
|
caller_agent=current["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Complete after the same cold pause. FAKE_SLEEP_SECONDS=4",
|
|
mode="read-only",
|
|
)
|
|
worker_id = str(job["job_id"])
|
|
for _ in range(500):
|
|
worker = load_job(worker_id)
|
|
if isinstance(worker.get("active_turn_id"), str):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose an active turn")
|
|
worker_thread_id = str(worker["app_server_thread_id"])
|
|
old_worker_pid = int(worker["runner_pid"])
|
|
old_worker_token = str(worker["runner_start_token"])
|
|
initial_wait = wait_for_jobs([worker_id], session_id=session_id, timeout_seconds=0)
|
|
self.assertNotIn("results", initial_wait)
|
|
baseline = initial_wait["progress_revisions"]
|
|
with self.assertRaisesRegex(ValueError, "exactly the requested jobs"):
|
|
wait_for_jobs(
|
|
[worker_id],
|
|
session_id=session_id,
|
|
timeout_seconds=0,
|
|
after_revision={},
|
|
)
|
|
|
|
paused = pause_session(session_id)
|
|
self.assertEqual(paused["session"]["status"], "paused")
|
|
self.assertEqual(paused["affected_jobs"], [worker_id])
|
|
self.assertTrue(paused["session"]["runtime_current"])
|
|
self.assertFalse(process_matches(old_root_pid, old_root_token))
|
|
self.assertFalse(process_matches(old_worker_pid, old_worker_token))
|
|
paused_worker = load_job(worker_id)
|
|
self.assertEqual(paused_worker["status"], "paused")
|
|
self.assertIsNone(paused_worker.get("goal_status"))
|
|
self.assertNotIn("cold_pause_pending", paused_worker)
|
|
self.assertTrue(Path(paused_worker["partial_result_path"]).is_file())
|
|
self.assertTrue(mmo_runtime.public_job(paused_worker)["runtime_current"])
|
|
self.assertNotIn("cold_pause_pending", load_session(session_id))
|
|
changed = wait_for_jobs(
|
|
[worker_id],
|
|
session_id=session_id,
|
|
timeout_seconds=10,
|
|
after_revision=baseline,
|
|
)
|
|
self.assertEqual(changed["changed_job_ids"], [worker_id])
|
|
self.assertFalse(changed["timed_out_waiting"])
|
|
|
|
compacted = compact_session(session_id)
|
|
self.assertEqual(compacted["session"]["status"], "paused")
|
|
after_compact = load_session(session_id)
|
|
self.assertEqual(after_compact["root_thread_id"], root_thread_id)
|
|
self.assertIsInstance(after_compact.get("root_compacted_at"), str)
|
|
self.assertNotIn("root_pid", after_compact)
|
|
|
|
continued = continue_session(
|
|
session_id,
|
|
input_text="Continue from retained context after compaction.",
|
|
)
|
|
self.assertEqual(continued["resume_errors"], {})
|
|
self.assertEqual(continued["resumed_jobs"], [worker_id])
|
|
resumed_worker = load_job(worker_id)
|
|
self.assertEqual(resumed_worker["app_server_thread_id"], worker_thread_id)
|
|
|
|
waited = wait_for_jobs([worker_id], session_id=session_id, timeout_seconds=20)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
root_thread.join(timeout=20)
|
|
self.assertFalse(root_thread.is_alive())
|
|
self.assertFalse(root_failures, root_failures)
|
|
self.assertIn(root_results[0]["status"], {"completed", "detached"})
|
|
self.assertEqual(root_results[0]["session"]["root_thread_id"], root_thread_id)
|
|
deadline = time.monotonic() + 15
|
|
while load_session(session_id)["status"] not in (
|
|
mmo_runtime.TERMINAL_SESSION_STATUSES
|
|
):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(session_id)
|
|
self.assertEqual(completed["status"], "completed", completed)
|
|
self.assertEqual(completed["root_thread_id"], root_thread_id)
|
|
finally:
|
|
if session_id is not None:
|
|
current_session = load_session(session_id)
|
|
if current_session["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
cancel_session(session_id)
|
|
root_thread.join(timeout=10)
|
|
|
|
def test_smoke_tool_evidence_requires_successful_direct_calls(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
events = box.root / "smoke-events.jsonl"
|
|
events.write_text(
|
|
"\n".join(
|
|
json.dumps(row)
|
|
for row in (
|
|
{
|
|
"type": "item.completed",
|
|
"item": {
|
|
"type": "mcp_tool_call",
|
|
"server": "ida",
|
|
"tool": "decompile",
|
|
"status": "completed",
|
|
"error": None,
|
|
},
|
|
},
|
|
{
|
|
"type": "item.completed",
|
|
"item": {
|
|
"type": "mcp_tool_call",
|
|
"server": "ida",
|
|
"tool": "disasm",
|
|
"status": "failed",
|
|
"error": {"message": "failed"},
|
|
},
|
|
},
|
|
)
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
self.assertEqual(_successful_mcp_tools(events), {"ida.decompile"})
|
|
|
|
def test_smoke_tool_evidence_accepts_successful_switchyard_bridge_calls(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
events = box.root / "bridged-smoke-events.jsonl"
|
|
events.write_text(
|
|
json.dumps(
|
|
{
|
|
"type": "item.completed",
|
|
"schema": {"type": {"unexpected": "object"}},
|
|
"item": {
|
|
"type": "dynamicToolCall",
|
|
"tool": "mmo_mcp__repo_search__query",
|
|
"status": "completed",
|
|
"success": True,
|
|
"arguments": {"query": "bounded"},
|
|
},
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
required = ["repo.search.query"]
|
|
self.assertEqual(
|
|
_successful_mcp_tools(
|
|
events,
|
|
dynamic_tool_aliases=_dynamic_tool_aliases(required),
|
|
),
|
|
{"repo.search.query"},
|
|
)
|
|
|
|
def test_route_telemetry_distinguishes_selected_provider_fallbacks_and_retries(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
events = box.root / "route-events.jsonl"
|
|
events.write_text(
|
|
json.dumps(
|
|
{
|
|
"routing": {
|
|
"serving_provider_slug": "parasail",
|
|
"serving_endpoint_tag": "parasail/fp8",
|
|
"attempt": 2,
|
|
"retry_count": 1,
|
|
"fallback_index": 0,
|
|
}
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
metadata = {
|
|
"requested_route_policy": {
|
|
"only": ["parasail/fp8"],
|
|
"allow_fallbacks": False,
|
|
},
|
|
"model_key": "openrouter_openai_chat__deepseek_deepseek_v4_pro",
|
|
"model": "deepseek/deepseek-v4-pro",
|
|
"route": "openrouter_openai_chat",
|
|
}
|
|
observed = route_telemetry(metadata, {}, events)
|
|
self.assertTrue(observed["complete"])
|
|
self.assertEqual(observed["actual_serving_provider_slugs"], ["parasail"])
|
|
self.assertEqual(observed["actual_serving_endpoint_tags"], ["parasail/fp8"])
|
|
self.assertEqual(observed["successful_attempt"], 2)
|
|
self.assertEqual(observed["retries"], 1)
|
|
self.assertEqual(observed["fallback_index"], 0)
|
|
self.assertTrue(observed["retry_telemetry_complete"])
|
|
|
|
events.write_text(
|
|
json.dumps(
|
|
{
|
|
"openrouter_metadata": {
|
|
"attempt": 2,
|
|
"endpoints": {
|
|
"available": [
|
|
{"provider": "First", "selected": False},
|
|
{"provider": "Chosen", "selected": True},
|
|
]
|
|
},
|
|
"attempts": [
|
|
{"provider": "First", "status": 529},
|
|
{"provider": "Chosen", "status": 200},
|
|
],
|
|
}
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
fallback = route_telemetry(metadata, {}, events)
|
|
self.assertTrue(fallback["complete"])
|
|
self.assertEqual(fallback["actual_serving_provider_slugs"], ["Chosen"])
|
|
self.assertEqual(fallback["successful_attempt"], 2)
|
|
self.assertEqual(fallback["fallback_index"], 1)
|
|
self.assertIsNone(fallback["retries"])
|
|
self.assertFalse(fallback["retry_telemetry_complete"])
|
|
|
|
events.write_text(
|
|
json.dumps({"routing": {"attempt": 2}}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
incomplete = route_telemetry(metadata, {}, events)
|
|
self.assertFalse(incomplete["complete"])
|
|
self.assertEqual(incomplete["actual_serving_provider_slugs"], [])
|
|
self.assertEqual(incomplete["actual_serving_endpoint_tags"], [])
|
|
self.assertEqual(incomplete["fallback_index"], 1)
|
|
self.assertIsNone(incomplete["retries"])
|
|
|
|
def test_session_can_be_created_from_an_existing_immutable_snapshot(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "snapshot-source"
|
|
shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
|
|
snapshot = compile_profile(profile)
|
|
snapshot_hash = snapshot["manifest"]["snapshot_hash"]
|
|
shutil.rmtree(profile)
|
|
session = create_session(cwd=box.workspace, snapshot_hash=snapshot_hash)
|
|
try:
|
|
self.assertEqual(session["snapshot_hash"], snapshot_hash)
|
|
self.assertEqual(session["profile_id"], "codex-harness-team")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
with self.assertRaisesRegex(ValueError, "cannot be combined"):
|
|
create_session(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
snapshot_hash=snapshot_hash,
|
|
)
|
|
|
|
def test_interactive_resume_reuses_logical_session_home_snapshot_and_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
self.assertEqual(
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace),
|
|
0,
|
|
)
|
|
first = iter_sessions()[0]
|
|
session_id = first["session_id"]
|
|
root_thread_id = first["root_thread_id"]
|
|
root_home = first["homes"][first["root_agent"]]["home"]
|
|
snapshot_hash = first["snapshot_hash"]
|
|
self.assertIsInstance(root_thread_id, str)
|
|
self.assertTrue(root_thread_id)
|
|
self.assertEqual(first["run_sequence"], 1)
|
|
run_id = first["current_run_id"]
|
|
self.assertIsInstance(run_id, str)
|
|
self.assertEqual(first["status"], "detached")
|
|
self.assertEqual(first["root_execution_host"], "app_server")
|
|
|
|
snapshot = load_snapshot(snapshot_hash)
|
|
expected_agents = (
|
|
Path(snapshot["directory"]) / agent_guidance_relative_path(first["root_agent"])
|
|
).read_text(encoding="utf-8")
|
|
expected_skill = (Path(snapshot["directory"]) / PROFILE_SKILL_RELATIVE_PATH).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
skill_path = Path(first["homes"][first["root_agent"]]["orchestration_skill"])
|
|
(Path(root_home) / "AGENTS.md").write_text("tampered\n", encoding="utf-8")
|
|
skill_path.write_text("tampered\n", encoding="utf-8")
|
|
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"compile_profile",
|
|
side_effect=AssertionError("resume must not compile the current profile"),
|
|
):
|
|
self.assertEqual(resume_interactive(session_id), 0)
|
|
second = load_session(session_id)
|
|
self.assertEqual(second["session_id"], session_id)
|
|
self.assertEqual(second["root_thread_id"], root_thread_id)
|
|
self.assertEqual(second["homes"][second["root_agent"]]["home"], root_home)
|
|
self.assertEqual(second["snapshot_hash"], snapshot_hash)
|
|
self.assertEqual(second["run_sequence"], 1)
|
|
self.assertEqual(second["current_run_id"], run_id)
|
|
self.assertEqual(second["status"], "detached")
|
|
self.assertEqual(second["root_execution_host"], "app_server")
|
|
self.assertEqual(len(iter_sessions()), 1)
|
|
self.assertEqual(
|
|
(Path(root_home) / "AGENTS.md").read_text(encoding="utf-8"), expected_agents
|
|
)
|
|
self.assertEqual(skill_path.read_text(encoding="utf-8"), expected_skill)
|
|
self.assertEqual(
|
|
second["homes"][second["root_agent"]]["orchestration_skill_sha256"],
|
|
sha256_file(skill_path),
|
|
)
|
|
runs = iter_session_runs(session_id)
|
|
self.assertEqual(
|
|
[(item["sequence"], item["kind"], item["status"]) for item in runs],
|
|
[(1, "initial", "detached")],
|
|
)
|
|
invocations = [
|
|
json.loads(line)
|
|
for line in (Path(root_home) / "fake-interactive-invocations.jsonl")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
]
|
|
self.assertEqual(len(invocations), 2)
|
|
self.assertEqual(invocations[1]["args"][-2:], ["resume", root_thread_id])
|
|
self.assertIn("--remote", invocations[1]["args"])
|
|
self.assertEqual(invocations[1]["thread_id"], root_thread_id)
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_goal_interactive_session_is_seeded_paused_without_false_suspension(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
self.assertEqual(
|
|
launch_interactive(profile="adaptive-engineering", cwd=box.workspace),
|
|
0,
|
|
)
|
|
first = iter_sessions()[0]
|
|
session_id = first["session_id"]
|
|
started_at = first["started_at"]
|
|
thread_id = first["root_thread_id"]
|
|
self.assertEqual(first["status"], "detached")
|
|
self.assertEqual(first["root_goal_status"], "paused")
|
|
self.assertTrue(first["root_goal_bootstrap_pending"])
|
|
self.assertIsInstance(first["root_goal_objective"], str)
|
|
self.assertLessEqual(len(first["root_goal_objective"]), 4000)
|
|
|
|
self.assertEqual(resume_interactive(session_id), 0)
|
|
second = load_session(session_id)
|
|
self.assertEqual(second["status"], "detached")
|
|
self.assertEqual(second["root_goal_status"], "paused")
|
|
self.assertTrue(second["root_goal_bootstrap_pending"])
|
|
self.assertEqual(second["root_thread_id"], thread_id)
|
|
self.assertEqual(second["started_at"], started_at)
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_materialization_rechecks_the_guidance_bytes_it_reads(self) -> None:
|
|
with RuntimeSandbox():
|
|
snapshot = compile_profile("codex-harness-team")
|
|
cached = load_snapshot(snapshot["manifest"]["snapshot_hash"])
|
|
skill = Path(snapshot["directory"]) / PROFILE_SKILL_RELATIVE_PATH
|
|
skill.chmod(0o600)
|
|
original = skill.read_text(encoding="utf-8")
|
|
replacement = ("X" if original[0] != "X" else "Y") + original[1:]
|
|
self.assertEqual(len(replacement), len(original))
|
|
skill.write_text(replacement, encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "failed integrity validation"):
|
|
_snapshot_guidance_text(cached, PROFILE_SKILL_RELATIVE_PATH)
|
|
|
|
def test_resume_preserves_immutable_run_capabilities_and_mcp_identity(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
first = load_session(session_id)
|
|
run_id = first["current_run_id"]
|
|
old_token = root_mcp_token(session_id)
|
|
self.assertEqual(resume_interactive(session_id), 0)
|
|
second = load_session(session_id)
|
|
self.assertEqual(second["current_run_id"], run_id)
|
|
self.assertEqual(second["run_sequence"], 1)
|
|
self.assertEqual(root_mcp_token(session_id), old_token)
|
|
self.assertEqual(
|
|
load_session_run(session_id, run_id)["root_mcp_token_hash"],
|
|
first["root_mcp_token_hash"],
|
|
)
|
|
exchange = mcp_exchange(
|
|
second,
|
|
second["root_agent"],
|
|
caller_token=old_token,
|
|
run_id=run_id,
|
|
)
|
|
self.assertEqual(exchange.returncode, 0, exchange.stderr)
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_root_capability_file_is_the_only_runtime_source_of_truth(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = session["session_id"]
|
|
self.assertTrue(root_mcp_token(session_id))
|
|
capability_path = box.state / "sessions" / session_id / "capabilities.json"
|
|
capability_path.unlink()
|
|
with self.assertRaisesRegex(RuntimeError, "capability is unavailable"):
|
|
root_mcp_token(session_id)
|
|
finish_session(session_id, exit_code=1, error="test cleanup")
|
|
|
|
def test_late_launcher_publication_cannot_close_a_detached_immutable_run(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
detached = load_session(session_id)
|
|
run_id = detached["current_run_id"]
|
|
late_finish = finish_session(
|
|
session_id,
|
|
exit_code=0,
|
|
expected_run_id=run_id,
|
|
)
|
|
self.assertEqual(late_finish["status"], "detached")
|
|
self.assertEqual(late_finish["current_run_id"], run_id)
|
|
self.assertEqual(load_session(session_id)["root_thread_id"], detached["root_thread_id"])
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_cancellation_waits_for_the_immutable_session_lifecycle_lock(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
cancellation: list[dict[str, Any]] = []
|
|
directory = mmo_runtime.session_dir(session_id)
|
|
with mmo_runtime.file_lock(mmo_runtime.session_lifecycle_lock_path(directory)):
|
|
cancel_thread = threading.Thread(
|
|
target=lambda: cancellation.append(cancel_session(session_id)), daemon=True
|
|
)
|
|
cancel_thread.start()
|
|
time.sleep(0.2)
|
|
self.assertTrue(cancel_thread.is_alive())
|
|
self.assertEqual(load_session(session_id)["status"], "cancelling")
|
|
cancel_thread.join(10)
|
|
self.assertFalse(cancel_thread.is_alive())
|
|
self.assertEqual(cancellation[0]["session"]["status"], "cancelled")
|
|
with self.assertRaisesRegex(RuntimeError, "terminal immutable session"):
|
|
begin_resume_run(session_id)
|
|
|
|
def test_cancellation_terminalizes_the_same_run_without_replacement(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
before = load_session(session_id)
|
|
run_id = before["current_run_id"]
|
|
cancelled = cancel_session(session_id)
|
|
self.assertEqual(cancelled["session"]["status"], "cancelled")
|
|
terminal = load_session(session_id)
|
|
self.assertIsNone(terminal["current_run_id"])
|
|
self.assertEqual(terminal["last_run_id"], run_id)
|
|
self.assertEqual(load_session_run(session_id, run_id)["status"], "cancelled")
|
|
self.assertEqual(len(iter_session_runs(session_id)), 1)
|
|
|
|
def test_invalid_lifecycle_lock_does_not_partially_cancel_session(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
session_directory = box.state / "sessions" / session["session_id"]
|
|
lifecycle_lock = session_directory / "lifecycle.lock"
|
|
lifecycle_lock.symlink_to(box.root / "outside-lifecycle.lock")
|
|
with self.assertRaisesRegex(RuntimeError, "lifecycle lock cannot traverse"):
|
|
cancel_session(session["session_id"])
|
|
self.assertEqual(load_session(session["session_id"])["status"], "starting")
|
|
lifecycle_lock.unlink()
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_resume_rejects_an_attached_client_and_symlinked_run_storage(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session = iter_sessions()[0]
|
|
session_id = session["session_id"]
|
|
attached = begin_resume_run(session_id)
|
|
self.assertEqual(attached["status"], "running")
|
|
with self.assertRaisesRegex(RuntimeError, "already has an attached"):
|
|
begin_resume_run(session_id)
|
|
detach_session(session_id)
|
|
|
|
session_directory = box.state / "sessions" / session_id
|
|
runs = session_directory / "runs"
|
|
run_id = str(load_session(session_id)["current_run_id"])
|
|
run_bytes = (runs / run_id / "run.json").read_bytes()
|
|
shutil.rmtree(runs)
|
|
with self.assertRaises((FileNotFoundError, RuntimeError)):
|
|
begin_resume_run(session_id)
|
|
outside = box.root / "outside-run-storage"
|
|
outside.mkdir()
|
|
runs.symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaises((ValueError, RuntimeError)):
|
|
begin_resume_run(session_id)
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
runs.unlink()
|
|
(runs / run_id).mkdir(parents=True)
|
|
(runs / run_id / "run.json").write_bytes(run_bytes)
|
|
cancel_session(session_id)
|
|
|
|
def test_persistent_records_reject_boolean_schema_versions_and_sequences(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
run_path = (
|
|
box.state
|
|
/ "sessions"
|
|
/ session["session_id"]
|
|
/ "runs"
|
|
/ session["current_run_id"]
|
|
/ "run.json"
|
|
)
|
|
run = json.loads(run_path.read_text(encoding="utf-8"))
|
|
run["sequence"] = True
|
|
run_path.write_text(json.dumps(run) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(ValueError, "sequence must be exactly 1"):
|
|
load_session_run(session["session_id"], session["current_run_id"])
|
|
|
|
run["sequence"] = 2
|
|
run_path.write_text(json.dumps(run) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(ValueError, "sequence must be exactly 1"):
|
|
load_session_run(session["session_id"], session["current_run_id"])
|
|
|
|
run["sequence"] = 1
|
|
run["kind"] = "resume"
|
|
run_path.write_text(json.dumps(run) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(ValueError, "run kind must be initial"):
|
|
load_session_run(session["session_id"], session["current_run_id"])
|
|
|
|
run["kind"] = "initial"
|
|
run_path.write_text(json.dumps(run) + "\n", encoding="utf-8")
|
|
|
|
session_path = box.state / "sessions" / session["session_id"] / "session.json"
|
|
current = json.loads(session_path.read_text(encoding="utf-8"))
|
|
session_path.write_text(
|
|
json.dumps({**current, "root_execution_host": "legacy_exec"}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "root execution host is invalid"):
|
|
load_session(session["session_id"])
|
|
|
|
session_path.write_text(
|
|
json.dumps({**current, "root_execution_policy_enforced": 1}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "retired fields"):
|
|
load_session(session["session_id"])
|
|
|
|
session_path.write_text(
|
|
json.dumps({**current, "schema_version": True}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "unsupported session state schema"):
|
|
load_session(session["session_id"])
|
|
session_path.write_text(json.dumps(current) + "\n", encoding="utf-8")
|
|
|
|
def test_resume_rejects_orphaned_replacement_run_inventory(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session = iter_sessions()[0]
|
|
orphan_id = "run-000002-orphaned"
|
|
orphan_directory = box.state / "sessions" / session["session_id"] / "runs" / orphan_id
|
|
orphan_directory.mkdir()
|
|
original = load_session_run(session["session_id"], session["current_run_id"])
|
|
(orphan_directory / "run.json").write_text(
|
|
json.dumps(
|
|
{
|
|
**original,
|
|
"run_id": orphan_id,
|
|
"sequence": 1,
|
|
"kind": "initial",
|
|
"status": "starting",
|
|
"created_at": mmo_runtime.utc_now(),
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "exactly its one active run"):
|
|
begin_resume_run(session["session_id"])
|
|
shutil.rmtree(orphan_directory)
|
|
resumed = begin_resume_run(session["session_id"])
|
|
self.assertEqual(resumed["run_sequence"], 1)
|
|
self.assertEqual(resumed["current_run_id"], session["current_run_id"])
|
|
detach_session(session["session_id"])
|
|
stop_session(session["session_id"], grace_seconds=0)
|
|
|
|
def test_resume_audit_failure_does_not_destroy_the_immutable_run(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"append_audit",
|
|
side_effect=OSError("synthetic audit write failure"),
|
|
):
|
|
with self.assertRaisesRegex(OSError, "synthetic audit write failure"):
|
|
begin_resume_run(session_id)
|
|
failed = load_session(session_id)
|
|
self.assertEqual(failed["status"], "running")
|
|
self.assertIsInstance(failed["current_run_id"], str)
|
|
self.assertIsNone(failed["last_run_id"])
|
|
detach_session(session_id)
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_initial_audit_failure_terminalizes_the_created_session(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"append_audit",
|
|
side_effect=OSError("synthetic initial audit failure"),
|
|
):
|
|
with self.assertRaisesRegex(OSError, "synthetic initial audit failure"):
|
|
create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
failed = iter_sessions()[0]
|
|
self.assertEqual(failed["status"], "failed")
|
|
self.assertIsNone(failed["current_run_id"])
|
|
self.assertEqual(
|
|
load_session_run(failed["session_id"], failed["last_run_id"])["status"],
|
|
"failed",
|
|
)
|
|
|
|
def test_resume_fails_closed_on_generated_model_catalog_drift(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="adaptive-engineering", cwd=box.workspace)
|
|
session = iter_sessions()[0]
|
|
catalog_path = Path(session["homes"]["implementation_specialist"]["model_catalog_json"])
|
|
self.assertEqual(
|
|
session["homes"]["implementation_specialist"]["model_catalog_sha256"],
|
|
sha256_file(catalog_path),
|
|
)
|
|
with catalog_path.open("a", encoding="utf-8") as handle:
|
|
handle.write("\n")
|
|
with self.assertRaisesRegex(RuntimeError, "model catalog failed integrity"):
|
|
begin_resume_run(session["session_id"])
|
|
failed = load_session(session["session_id"])
|
|
self.assertEqual(failed["status"], "suspended")
|
|
self.assertEqual(failed["run_sequence"], 1)
|
|
self.assertEqual(failed["current_run_id"], session["current_run_id"])
|
|
self.assertIn("home validation failed", failed["resume_error"])
|
|
cancel_session(session["session_id"])
|
|
|
|
def test_resume_uses_thread_identity_without_rollout_file_discovery(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session = iter_sessions()[0]
|
|
home = Path(session["homes"][session["root_agent"]]["home"])
|
|
rollouts = list((home / "sessions").rglob(f"*{session['root_thread_id']}.jsonl"))
|
|
self.assertEqual(len(rollouts), 1)
|
|
duplicate = rollouts[0].with_name("duplicate-" + rollouts[0].name)
|
|
shutil.copy2(rollouts[0], duplicate)
|
|
resumed = begin_resume_run(session["session_id"])
|
|
self.assertEqual(resumed["root_thread_id"], session["root_thread_id"])
|
|
self.assertNotIn("root_rollout_path", resumed)
|
|
detach_session(session["session_id"])
|
|
stop_session(session["session_id"], grace_seconds=0)
|
|
|
|
def test_resume_selection_is_exact_cwd_scoped_and_rejects_noncanonical_sessions(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session = iter_sessions()[0]
|
|
self.assertEqual(resolve_resume_session(session["session_id"]), session["session_id"])
|
|
self.assertEqual(
|
|
resolve_resume_session(session["root_thread_id"]),
|
|
session["session_id"],
|
|
)
|
|
duplicate = launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
self.assertEqual(duplicate, 0)
|
|
other = next(
|
|
item for item in iter_sessions() if item["session_id"] != session["session_id"]
|
|
)
|
|
update_session(
|
|
other["session_id"],
|
|
**root_thread_binding(session["root_thread_id"]),
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "identifies multiple MMO sessions"):
|
|
resolve_resume_session(session["root_thread_id"])
|
|
update_session(
|
|
other["session_id"],
|
|
**root_thread_binding(other["root_thread_id"]),
|
|
)
|
|
stop_session(other["session_id"], grace_seconds=0)
|
|
self.assertEqual(
|
|
resolve_resume_session(last=True, cwd=box.workspace),
|
|
session["session_id"],
|
|
)
|
|
elsewhere = box.root / "elsewhere"
|
|
elsewhere.mkdir()
|
|
with self.assertRaisesRegex(FileNotFoundError, "no resumable"):
|
|
resolve_resume_session(last=True, cwd=elsewhere)
|
|
self.assertEqual(
|
|
resolve_resume_session(last=True, all_cwds=True, cwd=elsewhere),
|
|
session["session_id"],
|
|
)
|
|
root_thread_id = session["root_thread_id"]
|
|
update_session(
|
|
session["session_id"],
|
|
resume_error="synthetic wrapper crash before thread binding",
|
|
**root_thread_binding(None),
|
|
)
|
|
with self.assertRaisesRegex(FileNotFoundError, "no resumable"):
|
|
resolve_resume_session(last=True, cwd=box.workspace)
|
|
update_session(
|
|
session["session_id"],
|
|
**root_thread_binding(root_thread_id),
|
|
)
|
|
session_path = box.state / "sessions" / session["session_id"] / "session.json"
|
|
restored_state = json.loads(session_path.read_text(encoding="utf-8"))
|
|
session_path.write_text(
|
|
json.dumps({**restored_state, "schema_version": None}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "unsupported session state schema"):
|
|
begin_resume_run(session["session_id"])
|
|
# Restore the deliberately corrupted record without invoking the
|
|
# strict state reader that the preceding assertion is exercising.
|
|
restored_state["schema_version"] = mmo_runtime.MMO_SCHEMA_VERSION
|
|
session_path.write_text(
|
|
json.dumps(restored_state, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
corrupt = box.state / "sessions" / "corrupt-session"
|
|
corrupt.mkdir()
|
|
(corrupt / "session.json").write_text("{}\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "invalid session state"):
|
|
resolve_resume_session(last=True, cwd=box.workspace)
|
|
shutil.rmtree(corrupt)
|
|
stop_session(session["session_id"], grace_seconds=0)
|
|
|
|
def test_tainted_resume_requires_acknowledgement_and_disables_native_delegation(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
taint_session(session_id, "synthetic persistent boundary failure")
|
|
taint_events = [
|
|
json.loads(line)
|
|
for line in Path(load_session(session_id)["audit_path"])
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
if '"event":"session_tainted"' in line
|
|
]
|
|
self.assertEqual(taint_events[-1]["run_id"], load_session(session_id)["current_run_id"])
|
|
with self.assertRaisesRegex(RuntimeError, "--allow-tainted"):
|
|
begin_resume_run(session_id)
|
|
resumed = begin_resume_run(session_id, allow_tainted=True)
|
|
try:
|
|
root_home = Path(resumed["homes"][resumed["root_agent"]]["home"])
|
|
config = tomllib.loads((root_home / "config.toml").read_text(encoding="utf-8"))
|
|
self.assertFalse(config["agents"]["enabled"])
|
|
exchange = mcp_exchange(resumed, resumed["root_agent"])
|
|
rows = [json.loads(line) for line in exchange.stdout.splitlines()]
|
|
names = {
|
|
item["name"]
|
|
for row in rows
|
|
if row.get("id") == 2 and "result" in row
|
|
for item in row["result"]["tools"]
|
|
}
|
|
self.assertNotIn("agent_spawn", names)
|
|
self.assertNotIn("agents_spawn", names)
|
|
self.assertNotIn("agent_patch_integrate", names)
|
|
finally:
|
|
detach_session(session_id)
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_terminal_worker_capacity_remains_reusable_after_resume(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="access-efficient-escalation-lab", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
session = load_session(session_id)
|
|
job = spawn_job(
|
|
session_id=session_id,
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="test",
|
|
task="Run one bounded read-only fake task to verify active-capacity accounting.",
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session_id,
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
self.assertEqual(resume_interactive(session_id), 0)
|
|
resumed = load_session(session_id)
|
|
second = spawn_job(
|
|
session_id=session_id,
|
|
caller_agent=resumed["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="test",
|
|
task="Run a second task after reattaching the same immutable run.",
|
|
mode="read-only",
|
|
)
|
|
second_wait = wait_for_jobs(
|
|
[second["job_id"]],
|
|
session_id=session_id,
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(second_wait["unfinished"], second_wait)
|
|
self.assertEqual(resumed["run_sequence"], 1)
|
|
self.assertEqual(resumed["current_run_id"], session["current_run_id"])
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_session_retention_uses_latest_resume_activity(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
launch_interactive(profile="codex-harness-team", cwd=box.workspace)
|
|
session_id = iter_sessions()[0]["session_id"]
|
|
old = "2000-01-01T00:00:00.000+00:00"
|
|
recent = mmo_runtime.utc_now()
|
|
update_session(
|
|
session_id,
|
|
created_at=old,
|
|
finished_at=old,
|
|
last_active_at=recent,
|
|
)
|
|
self.assertEqual(
|
|
clean_state(job_days=0, session_days=1),
|
|
{"jobs": 0, "sessions": 0},
|
|
)
|
|
update_session(session_id, last_active_at=old)
|
|
self.assertEqual(
|
|
clean_state(job_days=0, session_days=1),
|
|
{"jobs": 0, "sessions": 0},
|
|
)
|
|
stop_session(session_id, grace_seconds=0)
|
|
update_session(session_id, finished_at=old, last_active_at=old)
|
|
self.assertEqual(
|
|
clean_state(job_days=0, session_days=1),
|
|
{"jobs": 0, "sessions": 1},
|
|
)
|
|
self.assertFalse((box.state / "sessions" / session_id).exists())
|
|
|
|
def test_missing_optional_route_credentials_degrade_only_those_roles(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
credentials = box.config / "credentials.env"
|
|
credentials.write_text(
|
|
"ZAI_CODING_API_KEY=fake-zai-coding\n"
|
|
"OPENROUTER_API_KEY=fake-openrouter\n"
|
|
"OPENAI_API_KEY=fake-openai\n",
|
|
encoding="utf-8",
|
|
)
|
|
with mock.patch.dict(os.environ, {"OPENCODE_API_KEY": ""}):
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
availability = session["route_availability"]
|
|
self.assertTrue(availability["codex_chatgpt_builtin"]["available"])
|
|
self.assertFalse(availability["opencode_go_openai_chat"]["available"])
|
|
self.assertFalse(availability["opencode_zen_anthropic_messages"]["available"])
|
|
with self.assertRaises(AdmissionError) as rejected:
|
|
spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="implement",
|
|
task="Implement one bounded branch after an optional route became unavailable.",
|
|
mode="read-only",
|
|
)
|
|
self.assertEqual(rejected.exception.reason, "route_unavailable")
|
|
self.assertIn("missing credential", str(rejected.exception))
|
|
self.assertEqual(list_jobs(session_id=session["session_id"]), [])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_typed_route_faults_are_immutable_and_exact(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
with self.assertRaisesRegex(ValueError, "must map"):
|
|
create_session(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
route_faults=cast(Any, []),
|
|
)
|
|
empty_faults = create_session(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
route_faults={},
|
|
)
|
|
self.assertEqual(empty_faults["route_faults"], {})
|
|
finish_session(empty_faults["session_id"], exit_code=0)
|
|
session = create_session(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
route_faults={"opencode_go_openai_chat": "rate_limit"},
|
|
)
|
|
try:
|
|
status = session["route_availability"]["opencode_go_openai_chat"]
|
|
self.assertFalse(status["available"])
|
|
self.assertEqual(status["fault"], "rate_limit")
|
|
self.assertEqual(status["reason"], "injected route fault: rate_limit")
|
|
self.assertEqual(
|
|
session["route_faults"],
|
|
{"opencode_go_openai_chat": "rate_limit"},
|
|
)
|
|
healthy = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
try:
|
|
self.assertTrue(
|
|
healthy["route_availability"]["opencode_go_openai_chat"]["available"]
|
|
)
|
|
self.assertEqual(session["gateway_pid"], healthy["gateway_pid"])
|
|
self.assertEqual(load_session(session["session_id"])["status"], "starting")
|
|
finally:
|
|
finish_session(healthy["session_id"], exit_code=0)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
with self.assertRaisesRegex(ValueError, "must be one of"):
|
|
create_session(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
route_faults={"opencode_go_openai_chat": "generic_unavailable"},
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "unknown route"):
|
|
create_session(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
route_faults={"provider": "timeout"},
|
|
)
|
|
|
|
def test_env_backed_provider_header_survives_secret_filter(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
fragment = {
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"routes": {
|
|
"header_route": {
|
|
"name": "Header-backed test route",
|
|
"driver": "codex_custom",
|
|
"api_operator": "example",
|
|
"access_product": "example_api",
|
|
"wire_protocol": "openai_responses",
|
|
"billing_mode": "api",
|
|
"base_url": "https://example.invalid/v1",
|
|
"transport_modalities": ["text", "image"],
|
|
"preserves_tool_media": True,
|
|
"tool_result_modalities": ["text", "image"],
|
|
"env_http_headers": {"X-Secret": "EXAMPLE_HEADER_SECRET"},
|
|
}
|
|
},
|
|
"models": {
|
|
"header_route__header_model": {
|
|
"route": "header_route",
|
|
"upstream_id": "header-model",
|
|
"maker": "example",
|
|
"modalities": ["text", "image"],
|
|
"reasoning_levels": ["high", "xhigh"],
|
|
"default_reasoning": "xhigh",
|
|
}
|
|
},
|
|
}
|
|
(box.config / "catalog.d" / "headers.toml").write_text(
|
|
toml_dumps(fragment), encoding="utf-8"
|
|
)
|
|
profile = box.root / "env-header-profile"
|
|
shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["agents"]["integrator"]["model"] = "header_route__header_model"
|
|
profile_data["agents"]["integrator"]["allowed_reasoning_efforts"] = [
|
|
"high",
|
|
"xhigh",
|
|
]
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
try:
|
|
root = session["root_agent"]
|
|
self.assertEqual(
|
|
session["homes"][root]["direct_header_envs"],
|
|
["EXAMPLE_HEADER_SECRET"],
|
|
)
|
|
with mock.patch.dict(os.environ, {"EXAMPLE_HEADER_SECRET": "header-secret-value"}):
|
|
environment = session_environment(session, root)
|
|
self.assertEqual(environment["EXAMPLE_HEADER_SECRET"], "header-secret-value")
|
|
|
|
credentials_path = box.config / "credentials.env"
|
|
credentials_path.write_text(
|
|
credentials_path.read_text(encoding="utf-8")
|
|
+ "EXAMPLE_HEADER_SECRET=file-header-secret\n",
|
|
encoding="utf-8",
|
|
)
|
|
with mock.patch.dict(os.environ, {"EXAMPLE_HEADER_SECRET": ""}):
|
|
environment = session_environment(session, root)
|
|
self.assertEqual(environment["EXAMPLE_HEADER_SECRET"], "file-header-secret")
|
|
|
|
with (
|
|
mock.patch.dict(os.environ, {"EXAMPLE_HEADER_SECRET": "bad\nvalue"}),
|
|
self.assertRaisesRegex(RuntimeError, "prohibited control character"),
|
|
):
|
|
session_environment(session, root)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_root_and_worker_image_commands_preserve_stdin_prompt_operand(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
image = box.workspace / "evidence.png"
|
|
image.write_bytes(b"fake image bytes accepted by the Codex stand-in")
|
|
root = run_root_exec(
|
|
profile="visual-engineering",
|
|
cwd=box.workspace,
|
|
prompt="Inspect the attached visual evidence.",
|
|
images=[str(image)],
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(root["status"], "completed", root)
|
|
|
|
session = create_session(profile="visual-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="visual_verifier",
|
|
task_kind="visual_verification",
|
|
task="Compare the attached visual evidence independently and report exact discrepancies.",
|
|
attachments=[str(image)],
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=True,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
finished = load_job(job["job_id"])
|
|
self.assertEqual(finished["status"], "completed", finished)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_successful_root_exec_retires_leftover_process_group(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Complete this deterministic test. FAKE_ORPHAN_CHILD",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(result["status"], "completed", result)
|
|
child_pid = int((box.workspace / "fake-orphan.pid").read_text(encoding="utf-8"))
|
|
self.assertFalse(process_alive(child_pid))
|
|
|
|
def test_successful_worker_retires_leftover_process_group(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="literal_scout",
|
|
literal_task={
|
|
"operation": "summarize_supplied",
|
|
"text": "FAKE_ORPHAN_CHILD",
|
|
"max_points": 3,
|
|
},
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
self.assertEqual(load_job(job["job_id"])["status"], "completed")
|
|
child_pid = int((box.workspace / "fake-orphan.pid").read_text(encoding="utf-8"))
|
|
self.assertFalse(process_alive(child_pid))
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_worker_host_retirement_reports_lingering_process_groups(self) -> None:
|
|
record = {
|
|
"runner_pid": 12345,
|
|
"runner_pgid": 12345,
|
|
"runner_start_token": "runner-token",
|
|
}
|
|
with (
|
|
mock.patch.object(mmo_runtime, "process_matches", return_value=True),
|
|
mock.patch.object(mmo_runtime, "process_group_alive", return_value=True),
|
|
mock.patch.object(mmo_runtime.os, "killpg") as killpg,
|
|
mock.patch.object(mmo_runtime.time, "monotonic", side_effect=[0.0, 0.0, 0.0, 3.0]),
|
|
mock.patch.object(mmo_runtime, "_reap_tracked_runner"),
|
|
self.assertRaisesRegex(RuntimeError, "process groups did not terminate"),
|
|
):
|
|
mmo_runtime._terminate_job_hosts([record], grace_seconds=0.0)
|
|
self.assertEqual(killpg.call_count, 2)
|
|
|
|
def test_runtime_digest_failure_does_not_break_state_inspection(self) -> None:
|
|
session = {
|
|
"root_runtime_package_version": mmo_runtime.package_version(),
|
|
"root_runtime_sha256": "unreadable",
|
|
}
|
|
job = {
|
|
"worker_runtime_package_version": mmo_runtime.package_version(),
|
|
"worker_runtime_sha256": "unreadable",
|
|
}
|
|
with mock.patch.object(Path, "read_bytes", side_effect=PermissionError("denied")):
|
|
self.assertFalse(mmo_runtime.public_session(session)["runtime_current"])
|
|
self.assertFalse(mmo_runtime.public_job(job)["runtime_current"])
|
|
|
|
def test_control_timeout_preserves_the_configured_lifecycle_window(self) -> None:
|
|
self.assertEqual(
|
|
mmo_runtime._root_control_timeout({"root_app_server_lifecycle_timeout_seconds": 1200}),
|
|
1230.0,
|
|
)
|
|
|
|
def test_interrupted_worker_cold_pause_retires_host_and_retains_evidence(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(
|
|
profile="incident-hypothesis-triage",
|
|
cwd=box.workspace,
|
|
)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Retain evidence while the pause controller disappears. FAKE_SLEEP_SECONDS=20",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(500):
|
|
current = load_job(job["job_id"])
|
|
if isinstance(current.get("active_turn_id"), str) and process_matches(
|
|
current.get("runner_pid"), current.get("runner_start_token")
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose a live turn")
|
|
old_pid = int(current["runner_pid"])
|
|
old_token = str(current["runner_start_token"])
|
|
directory = mmo_state.job_dir(job["job_id"])
|
|
with file_lock(mmo_state.runtime_lock_path()):
|
|
pending = read_job_record(directory)
|
|
pending.update(
|
|
cold_pause_pending=True,
|
|
last_control_status="delivery_unknown",
|
|
)
|
|
publish_job_record(directory, pending)
|
|
|
|
paused = load_job(job["job_id"])
|
|
self.assertEqual(paused["status"], "paused")
|
|
self.assertNotIn("cold_pause_pending", paused)
|
|
self.assertNotIn("runner_pid", paused)
|
|
self.assertFalse(process_matches(old_pid, old_token))
|
|
self.assertTrue(Path(paused["partial_result_path"]).is_file())
|
|
finally:
|
|
worker = load_job(job["job_id"])
|
|
if worker["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
current = load_session(session["session_id"])
|
|
if current["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_root_exec_cannot_escalate_a_read_only_profile_root(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "read-only-root"
|
|
shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["agents"]["integrator"]["permissions"] = "read-only"
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
with self.assertRaisesRegex(PermissionError, "permanently read-only") as failure:
|
|
run_root_exec(
|
|
profile=profile,
|
|
cwd=box.workspace,
|
|
prompt="Attempt a write-capable root execution.",
|
|
sandbox_mode="workspace-write",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(getattr(failure.exception, "mmo_session_status", None), "failed")
|
|
self.assertIsInstance(getattr(failure.exception, "mmo_session_id", None), str)
|
|
sessions = iter_sessions()
|
|
self.assertEqual(len(sessions), 1)
|
|
self.assertEqual(sessions[0]["status"], "failed")
|
|
|
|
def test_interactive_client_failure_detaches_without_destroying_the_host(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
created: list[dict] = []
|
|
|
|
def capture_session(
|
|
*,
|
|
profile: str | Path | None = None,
|
|
cwd: str | Path | None = None,
|
|
bindings: Mapping[str, str] | None = None,
|
|
session_kind: str = "noninteractive",
|
|
) -> dict[str, Any]:
|
|
session = create_session(
|
|
profile=profile,
|
|
cwd=cwd,
|
|
bindings=bindings,
|
|
session_kind=session_kind,
|
|
)
|
|
created.append(session)
|
|
return session
|
|
|
|
with (
|
|
mock.patch.object(mmo_runtime, "create_session", side_effect=capture_session),
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"session_environment",
|
|
side_effect=RuntimeError("synthetic environment failure"),
|
|
),
|
|
self.assertRaisesRegex(RuntimeError, "synthetic environment failure"),
|
|
):
|
|
mmo_runtime.launch_interactive(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
)
|
|
self.assertEqual(len(created), 1)
|
|
session_id = created[0]["session_id"]
|
|
persisted = load_session(session_id)
|
|
self.assertEqual(persisted["status"], "detached")
|
|
self.assertIsInstance(persisted["root_thread_id"], str)
|
|
self.assertTrue(process_matches(persisted["root_pid"], persisted["root_start_token"]))
|
|
self.assertTrue(root_mcp_token(session_id))
|
|
stop_session(session_id, grace_seconds=0)
|
|
|
|
def test_gateway_and_generated_hybrid_native_configs(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
(box.config / "tool-mcp.d" / "servers.toml").write_text(
|
|
f"""schema_version = {mmo_runtime.MMO_SCHEMA_VERSION}
|
|
|
|
[tool_mcp_servers.firecrawl]
|
|
transport = "streamable_http"
|
|
url = "https://example.invalid/mcp"
|
|
bearer_token_env_var = "FIRECRAWL_API_KEY"
|
|
enabled_tools = ["search", "scrape"]
|
|
default_tools_approval_mode = "writes"
|
|
|
|
[tool_mcp_servers.ida_pro]
|
|
transport = "stdio"
|
|
command = "/bin/true"
|
|
env_vars = ["IDA_MCP_TOKEN"]
|
|
enabled_tools = ["inspect", "decompile"]
|
|
default_tools_approval_mode = "prompt"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
with (box.config / "credentials.env").open("a", encoding="utf-8") as credentials:
|
|
credentials.write("FIRECRAWL_API_KEY=firecrawl-secret\nIDA_MCP_TOKEN=ida-secret\n")
|
|
profile = box.root / "hybrid-tool-mcp"
|
|
shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["agents"]["orchestrator"]["tool_mcp_servers"] = {
|
|
"firecrawl": {"enabled_tools": ["search"]}
|
|
}
|
|
profile_data["agents"]["repo_scout"]["tool_mcp_servers"] = {
|
|
"firecrawl": {"required": False},
|
|
"ida_pro": {"enabled_tools": ["inspect"]},
|
|
}
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
try:
|
|
models = gateway_models(session["snapshot_hash"])
|
|
advertised = {row["id"] for row in models["data"]}
|
|
snapshot = compile_profile(profile)
|
|
self.assertTrue(set(snapshot["manifest"]["route_ids"].values()) <= advertised)
|
|
root_home = Path(session["homes"][session["root_agent"]]["home"])
|
|
root_config = tomllib.loads((root_home / "config.toml").read_text(encoding="utf-8"))
|
|
self.assertTrue(root_config["features"]["multi_agent"])
|
|
root_skill = Path(session["homes"][session["root_agent"]]["orchestration_skill"])
|
|
self.assertEqual(
|
|
root_skill,
|
|
root_home / "skills" / PROFILE_SKILL_NAME / "SKILL.md",
|
|
)
|
|
self.assertEqual(
|
|
root_config["skills"]["config"],
|
|
[{"path": str(root_skill), "enabled": True}],
|
|
)
|
|
self.assertEqual(
|
|
(root_home / "AGENTS.md").read_text(encoding="utf-8"),
|
|
(
|
|
Path(snapshot["directory"])
|
|
/ agent_guidance_relative_path(session["root_agent"])
|
|
).read_text(encoding="utf-8"),
|
|
)
|
|
self.assertEqual(
|
|
root_skill.read_text(encoding="utf-8"),
|
|
(Path(snapshot["directory"]) / PROFILE_SKILL_RELATIVE_PATH).read_text(
|
|
encoding="utf-8"
|
|
),
|
|
)
|
|
self.assertIn("mmo_mesh", root_config["mcp_servers"])
|
|
self.assertTrue(root_config["mcp_servers"]["firecrawl"]["enabled"])
|
|
self.assertEqual(
|
|
root_config["mcp_servers"]["firecrawl"]["enabled_tools"],
|
|
["search", "scrape"],
|
|
)
|
|
self.assertEqual(
|
|
root_config["mcp_servers"]["firecrawl"]["disabled_tools"],
|
|
["scrape"],
|
|
)
|
|
self.assertFalse(root_config["mcp_servers"]["ida_pro"]["enabled"])
|
|
self.assertEqual(
|
|
root_config["mcp_servers"]["mmo_mesh"]["env_vars"],
|
|
["MMO_CALLER_TOKEN", "MMO_RUN_ID"],
|
|
)
|
|
self.assertEqual(
|
|
root_config["mcp_servers"]["mmo_mesh"]["tool_timeout_sec"],
|
|
1230.0,
|
|
)
|
|
expected_mesh_tools = {
|
|
"agent_spawn",
|
|
"agents_spawn",
|
|
"agent_status",
|
|
"agents_wait",
|
|
"agent_result",
|
|
"agent_result_accept",
|
|
"agent_result_reject",
|
|
"agent_patch_integrate",
|
|
"agent_cancel",
|
|
} | set(AGENT_MCP_CONTROL_TOOLS)
|
|
self.assertEqual(
|
|
set(root_config["mcp_servers"]["mmo_mesh"]["enabled_tools"]),
|
|
expected_mesh_tools,
|
|
)
|
|
worker_home = Path(session["homes"]["implementation_specialist"]["home"])
|
|
worker_config = tomllib.loads(
|
|
(worker_home / "config.toml").read_text(encoding="utf-8")
|
|
)
|
|
worker_skill = Path(
|
|
session["homes"]["implementation_specialist"]["orchestration_skill"]
|
|
)
|
|
self.assertEqual(
|
|
worker_config["skills"]["config"],
|
|
[{"path": str(worker_skill), "enabled": True}],
|
|
)
|
|
self.assertIn("mmo_mesh", worker_config.get("mcp_servers", {}))
|
|
self.assertEqual(
|
|
worker_config["mcp_servers"]["mmo_mesh"]["tool_timeout_sec"],
|
|
1230.0,
|
|
)
|
|
self.assertEqual(
|
|
set(worker_config["mcp_servers"]["mmo_mesh"]["enabled_tools"]),
|
|
{
|
|
"agent_status",
|
|
"agents_wait",
|
|
"agent_result",
|
|
}
|
|
| set(AGENT_MCP_CONTROL_TOOLS),
|
|
)
|
|
self.assertNotIn(
|
|
"agent_result_accept",
|
|
worker_config["mcp_servers"]["mmo_mesh"]["enabled_tools"],
|
|
)
|
|
self.assertNotIn(
|
|
"agent_patch_integrate",
|
|
worker_config["mcp_servers"]["mmo_mesh"]["enabled_tools"],
|
|
)
|
|
self.assertNotIn(root_mcp_token(session["session_id"]), json.dumps(root_config))
|
|
self.assertNotIn("firecrawl-secret", json.dumps(root_config))
|
|
self.assertNotIn("ida-secret", json.dumps(root_config))
|
|
self.assertNotIn("model_context_window", root_config)
|
|
self.assertNotIn("model_supports_reasoning_summaries", root_config)
|
|
self.assertIn(
|
|
"repo_scout", session["homes"][session["root_agent"]]["native_agent_files"]
|
|
)
|
|
native_path = Path(
|
|
session["homes"][session["root_agent"]]["native_agent_files"]["repo_scout"]
|
|
)
|
|
native_config = tomllib.loads(native_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
native_config["skills"]["config"],
|
|
[{"path": str(root_skill), "enabled": False}],
|
|
)
|
|
self.assertNotIn("model_context_window", native_config)
|
|
self.assertNotIn("model_supports_reasoning_summaries", native_config)
|
|
self.assertFalse(native_config["sandbox_workspace_write"]["network_access"])
|
|
self.assertEqual(
|
|
native_config["mcp_servers"]["mmo_mesh"]["env_vars"],
|
|
["MMO_CALLER_TOKEN", "MMO_RUN_ID"],
|
|
)
|
|
self.assertEqual(
|
|
native_config["mcp_servers"]["mmo_mesh"]["tool_timeout_sec"],
|
|
1230.0,
|
|
)
|
|
self.assertTrue(native_config["mcp_servers"]["firecrawl"]["enabled"])
|
|
self.assertFalse(native_config["mcp_servers"]["firecrawl"]["required"])
|
|
self.assertEqual(native_config["mcp_servers"]["firecrawl"]["disabled_tools"], [])
|
|
self.assertTrue(native_config["mcp_servers"]["ida_pro"]["enabled"])
|
|
self.assertEqual(
|
|
native_config["mcp_servers"]["ida_pro"]["disabled_tools"],
|
|
["decompile"],
|
|
)
|
|
environment = session_environment(session, session["root_agent"])
|
|
self.assertEqual(environment["FIRECRAWL_API_KEY"], "firecrawl-secret")
|
|
self.assertEqual(environment["IDA_MCP_TOKEN"], "ida-secret")
|
|
self.assertEqual(
|
|
session["homes"][session["root_agent"]]["tool_mcp_envs"],
|
|
["FIRECRAWL_API_KEY", "IDA_MCP_TOKEN"],
|
|
)
|
|
self.assertEqual(
|
|
session["homes"][session["root_agent"]]["tool_mcp_http_envs"],
|
|
["FIRECRAWL_API_KEY"],
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
"tool MCP credentials" in warning for warning in session["profile_warnings"]
|
|
)
|
|
)
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"FIRECRAWL_API_KEY": "invalid\nheader"},
|
|
):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"tool MCP HTTP environment variable FIRECRAWL_API_KEY",
|
|
):
|
|
session_environment(session, session["root_agent"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_reasoning_none_is_not_emitted_to_codex(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
try:
|
|
qwen_home = Path(session["homes"]["literal_scout"]["home"])
|
|
qwen_config = tomllib.loads((qwen_home / "config.toml").read_text(encoding="utf-8"))
|
|
self.assertNotIn("model_reasoning_effort", qwen_config)
|
|
self.assertNotIn("plan_mode_reasoning_effort", qwen_config)
|
|
self.assertNotIn("model_supports_reasoning_summaries", qwen_config)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_cancelled_session_remains_cancelled_after_root_exit(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
cancelled = cancel_session(session["session_id"])
|
|
self.assertEqual(cancelled["session"]["status"], "cancelled")
|
|
finished = finish_session(session["session_id"], exit_code=0)
|
|
self.assertEqual(finished["status"], "cancelled")
|
|
self.assertEqual(load_session(session["session_id"])["status"], "cancelled")
|
|
|
|
def test_stale_starting_session_is_reconciled(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
update_session(
|
|
session["session_id"],
|
|
created_at="2000-01-01T00:00:00+00:00",
|
|
run_created_at="2000-01-01T00:00:00+00:00",
|
|
status="starting",
|
|
root_pid=None,
|
|
)
|
|
self.assertEqual(load_session(session["session_id"])["status"], "failed")
|
|
|
|
def test_process_identity_requires_matching_start_token(self) -> None:
|
|
self.assertFalse(process_matches(os.getpid(), None))
|
|
self.assertFalse(process_matches(os.getpid(), "not-the-current-start-token"))
|
|
|
|
def test_workspace_git_commands_never_read_interactive_stdin(self) -> None:
|
|
completed = subprocess.CompletedProcess([], 0, stdout=b"", stderr=b"")
|
|
with mock.patch.object(mmo_workspace.subprocess, "run", return_value=completed) as run:
|
|
self.assertIs(mmo_workspace._git(Path("/tmp"), "status"), completed)
|
|
self.assertIs(run.call_args.kwargs["stdin"], subprocess.DEVNULL)
|
|
|
|
def test_mcp_root_and_native_identity(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "native-identity-profile"
|
|
shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "native-identity-profile"
|
|
profile_data["agents"]["repo_scout"]["can_spawn"] = ["adversarial_reviewer"]
|
|
profile_data["agents"]["repo_scout"]["controls"] = {
|
|
"adversarial_reviewer": {
|
|
"actions": list(
|
|
profile_data["agents"]["orchestrator"]["controls"]["adversarial_reviewer"][
|
|
"actions"
|
|
]
|
|
)
|
|
}
|
|
}
|
|
profile_data["coordination"]["max_depth"] = 2
|
|
profile_data["coordination"]["native_nested_delegation"] = True
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
try:
|
|
root_result = mcp_exchange(session, session["root_agent"])
|
|
self.assertEqual(root_result.returncode, 0, root_result.stderr)
|
|
root_rows = [json.loads(line) for line in root_result.stdout.splitlines()]
|
|
tool_rows = [
|
|
item
|
|
for row in root_rows
|
|
if row.get("id") == 2
|
|
for item in row["result"]["tools"]
|
|
]
|
|
tools = [item["name"] for item in tool_rows]
|
|
self.assertEqual(
|
|
set(tools),
|
|
{
|
|
"agent_spawn",
|
|
"agents_spawn",
|
|
"agent_status",
|
|
"agents_wait",
|
|
"agent_result",
|
|
"agent_cancel",
|
|
"agent_result_accept",
|
|
"agent_result_reject",
|
|
"agent_patch_integrate",
|
|
}
|
|
| set(AGENT_MCP_CONTROL_TOOLS),
|
|
)
|
|
tools_by_name = {item["name"]: item for item in tool_rows}
|
|
spawn_schema = tools_by_name["agent_spawn"]["inputSchema"]
|
|
self.assertNotIn("oneOf", spawn_schema)
|
|
self.assertEqual(
|
|
spawn_schema["properties"]["agent"]["enum"],
|
|
[
|
|
"implementation_specialist",
|
|
"adversarial_reviewer",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
tools_by_name["agents_spawn"]["inputSchema"]["properties"]["agents"]["items"],
|
|
spawn_schema,
|
|
)
|
|
self.assertFalse(tools_by_name["agent_status"]["annotations"]["readOnlyHint"])
|
|
self.assertFalse(tools_by_name["agents_wait"]["annotations"]["readOnlyHint"])
|
|
self.assertFalse(tools_by_name["agent_result"]["annotations"]["readOnlyHint"])
|
|
self.assertTrue(
|
|
tools_by_name["agent_patch_integrate"]["annotations"]["destructiveHint"]
|
|
)
|
|
persisted = load_session(session["session_id"])
|
|
self.assertNotIn("root_mcp_token", persisted)
|
|
self.assertNotIn(root_mcp_token(session["session_id"]), json.dumps(persisted))
|
|
|
|
forged = mcp_exchange(session, "implementation_specialist")
|
|
self.assertIn("only the root role", forged.stdout)
|
|
wrong_capability = mcp_exchange(
|
|
session,
|
|
session["root_agent"],
|
|
caller_token="wrong",
|
|
)
|
|
self.assertIn("invalid MCP caller capability", wrong_capability.stdout)
|
|
|
|
native_path = Path(
|
|
session["homes"][session["root_agent"]]["native_agent_files"]["repo_scout"]
|
|
)
|
|
native_config = tomllib.loads(native_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
set(native_config["mcp_servers"]["mmo_mesh"]["enabled_tools"]),
|
|
set(tools),
|
|
)
|
|
token = native_config["mcp_servers"]["mmo_mesh"]["env"]["MMO_NATIVE_CALLER_TOKEN"]
|
|
valid = mcp_exchange(session, "repo_scout", native_token=token)
|
|
self.assertEqual(valid.returncode, 0, valid.stderr)
|
|
invalid = mcp_exchange(session, "repo_scout", native_token="wrong")
|
|
invalid_rows = [json.loads(line) for line in invalid.stdout.splitlines()]
|
|
self.assertTrue(any("error" in row for row in invalid_rows), invalid.stdout)
|
|
self.assertIn("invalid native-agent", invalid.stdout)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
with self.assertRaisesRegex(RuntimeError, "capability is unavailable"):
|
|
root_mcp_token(session["session_id"])
|
|
|
|
def test_nested_mcp_and_native_callers_receive_job_scoped_identity_without_progress(
|
|
self,
|
|
) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "nested-job-identity-profile"
|
|
shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "nested-job-identity-profile"
|
|
profile_data["agents"]["implementation_specialist"]["can_spawn"] = [
|
|
"repo_scout",
|
|
"adversarial_reviewer",
|
|
]
|
|
profile_data["agents"]["repo_scout"]["can_spawn"] = ["adversarial_reviewer"]
|
|
profile_data["coordination"]["max_depth"] = 3
|
|
profile_data["coordination"]["native_nested_delegation"] = True
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
worker_home = Path(session["homes"]["implementation_specialist"]["home"])
|
|
worker_config = tomllib.loads(
|
|
(worker_home / "config.toml").read_text(encoding="utf-8")
|
|
)
|
|
worker_skill = Path(
|
|
session["homes"]["implementation_specialist"]["orchestration_skill"]
|
|
)
|
|
self.assertEqual(
|
|
worker_config["skills"]["config"],
|
|
[{"path": str(worker_skill), "enabled": True}],
|
|
)
|
|
worker_mesh = worker_config["mcp_servers"]["mmo_mesh"]
|
|
self.assertEqual(
|
|
worker_mesh["env_vars"],
|
|
["MMO_CALLER_TOKEN", "MMO_RUN_ID", "MMO_CALLER_JOB_ID"],
|
|
)
|
|
self.assertNotIn("agent_progress", worker_mesh["enabled_tools"])
|
|
self.assertNotIn("agent_progress_history", worker_mesh["enabled_tools"])
|
|
|
|
native_path = Path(
|
|
session["homes"]["implementation_specialist"]["native_agent_files"][
|
|
"repo_scout"
|
|
]
|
|
)
|
|
native_config = tomllib.loads(native_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
native_config["skills"]["config"],
|
|
[{"path": str(worker_skill), "enabled": True}],
|
|
)
|
|
native_mesh = native_config["mcp_servers"]["mmo_mesh"]
|
|
self.assertEqual(
|
|
native_mesh["env_vars"],
|
|
["MMO_CALLER_TOKEN", "MMO_RUN_ID", "MMO_CALLER_JOB_ID"],
|
|
)
|
|
self.assertNotIn("agent_progress", native_mesh["enabled_tools"])
|
|
self.assertNotIn("agent_progress_history", native_mesh["enabled_tools"])
|
|
|
|
caller_token = "nested-worker-mcp-capability"
|
|
with mock.patch.object(
|
|
mmo_runtime.secrets, "token_urlsafe", return_value=caller_token
|
|
):
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Inspect nested identity without writing. FAKE_SLEEP_SECONDS=2",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(100):
|
|
current = load_job(job["job_id"])
|
|
if current["status"] == "running":
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("nested worker did not enter running state")
|
|
manager_result = mcp_exchange(
|
|
session,
|
|
"implementation_specialist",
|
|
caller_token=caller_token,
|
|
caller_job_id=job["job_id"],
|
|
)
|
|
self.assertEqual(manager_result.returncode, 0, manager_result.stderr)
|
|
native_result = mcp_exchange(
|
|
session,
|
|
"repo_scout",
|
|
caller_token=caller_token,
|
|
caller_job_id=job["job_id"],
|
|
native_token=native_mesh["env"]["MMO_NATIVE_CALLER_TOKEN"],
|
|
)
|
|
self.assertEqual(native_result.returncode, 0, native_result.stderr)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_mcp_negotiation_lifecycle_and_json_rpc_validation(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
try:
|
|
result = mcp_raw_exchange(
|
|
session,
|
|
session["root_agent"],
|
|
[
|
|
'{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}',
|
|
json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": "initialize-request",
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2099-01-01",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "1"},
|
|
},
|
|
}
|
|
),
|
|
'{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}',
|
|
'{"jsonrpc":"2.0","id":true,"method":"tools/list","params":{}}',
|
|
'{"jsonrpc":"2.0","id":null,"method":"tools/list","params":{}}',
|
|
'{"jsonrpc":"2.0","id":1.5,"method":"tools/list","params":{}}',
|
|
'{"jsonrpc":"1.0","id":3,"method":"tools/list","params":{}}',
|
|
'{"jsonrpc":"2.0","id":8,"method":"tools/list","params":[]}',
|
|
'{"jsonrpc":"2.0","id":4,"method":"tools/list","params":{}}',
|
|
"[]",
|
|
'{"jsonrpc":"2.0","id":5,"method":"ping","params":{"n":NaN}}',
|
|
'{"jsonrpc":"2.0","id":6,"method":"ping","method":"tools/list"}',
|
|
'{"jsonrpc":"2.0","id":7,"method":"\\ud800"}',
|
|
],
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
rows = [json.loads(line) for line in result.stdout.splitlines()]
|
|
self.assertEqual(rows[0]["error"]["code"], -32600)
|
|
self.assertEqual(rows[0]["error"]["message"], "server not initialized")
|
|
self.assertEqual(rows[1]["id"], "initialize-request")
|
|
self.assertEqual(rows[1]["result"]["protocolVersion"], "2025-06-18")
|
|
self.assertEqual(rows[2]["error"]["code"], -32600)
|
|
self.assertIsNone(rows[2]["id"])
|
|
self.assertIsNone(rows[3]["id"])
|
|
self.assertEqual(rows[3]["error"]["code"], -32600)
|
|
self.assertIsNone(rows[4]["id"])
|
|
self.assertEqual(rows[4]["error"]["code"], -32600)
|
|
self.assertEqual(rows[5]["id"], 3)
|
|
self.assertEqual(rows[5]["error"]["code"], -32600)
|
|
self.assertEqual(rows[6]["id"], 8)
|
|
self.assertEqual(rows[6]["error"]["code"], -32600)
|
|
self.assertEqual(rows[7]["id"], 4)
|
|
self.assertIn("tools", rows[7]["result"])
|
|
self.assertEqual(rows[8]["error"]["code"], -32600)
|
|
self.assertEqual(rows[9]["error"]["code"], -32700)
|
|
self.assertEqual(rows[10]["error"]["code"], -32700)
|
|
self.assertEqual(rows[11]["error"]["code"], -32700)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_batch_admission_is_atomic(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
with self.assertRaisesRegex(ValueError, "unsupported literal_task operation"):
|
|
spawn_jobs(
|
|
[
|
|
{
|
|
"agent": "literal_scout",
|
|
"literal_task": {
|
|
"operation": "summarize_supplied",
|
|
"text": "README.md literal evidence fixture",
|
|
},
|
|
},
|
|
{
|
|
"agent": "literal_scout",
|
|
"literal_task": {"operation": "architecture"},
|
|
},
|
|
],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
)
|
|
self.assertEqual(list_jobs(session_id=session["session_id"]), [])
|
|
|
|
batch = spawn_jobs(
|
|
[
|
|
{
|
|
"agent": "literal_scout",
|
|
"literal_task": {
|
|
"operation": "summarize_supplied",
|
|
"text": "README.md literal evidence fixture",
|
|
},
|
|
},
|
|
{
|
|
"agent": "flagship_escalation",
|
|
"task_kind": "analysis",
|
|
"task": "Analyze one directly evidenced README.md inconsistency.",
|
|
},
|
|
],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
)
|
|
self.assertTrue(batch["atomic"])
|
|
self.assertEqual(len(batch["accepted"]), 2)
|
|
self.assertEqual(
|
|
len({item["batch_id"] for item in batch["accepted"]}),
|
|
1,
|
|
)
|
|
identifiers = [item["job_id"] for item in batch["accepted"]]
|
|
waited = wait_for_jobs(
|
|
identifiers,
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_async_worker_contract_and_result_consumption(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="literal_scout",
|
|
literal_task={
|
|
"operation": "summarize_supplied",
|
|
"text": "README.md literal evidence fixture",
|
|
},
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]], session_id=session["session_id"], timeout_seconds=20
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertTrue(final["contract_valid"])
|
|
self.assertNotIn("--output-schema", final["command"])
|
|
result = read_result(job["job_id"], session_id=session["session_id"])
|
|
self.assertEqual(result["content_format"], "json")
|
|
self.assertEqual(result["content"]["operation"], "summarize_supplied")
|
|
self.assertRegex(result["content"]["input_sha256"], r"^[0-9a-f]{64}$")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_app_server_worker_exposes_controls_without_legacy_progress_surface(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
caller_token = "unconfigured-worker-mcp-capability"
|
|
with mock.patch.object(
|
|
mmo_runtime.secrets, "token_urlsafe", return_value=caller_token
|
|
):
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Inspect one bounded invariant without writing. FAKE_SLEEP_SECONDS=2",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(100):
|
|
current = load_job(job["job_id"])
|
|
if current["status"] == "running" and current.get("command"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not enter running state")
|
|
|
|
self.assertEqual(current["execution_mode"], "goal")
|
|
self.assertGreater(current["goal_token_budget"], 0)
|
|
self.assertGreaterEqual(
|
|
current["max_goal_token_budget"], current["goal_token_budget"]
|
|
)
|
|
self.assertNotIn("progress_interval_seconds", current)
|
|
self.assertNotIn("timeout_seconds", current)
|
|
self.assertNotIn("execution_policy", current)
|
|
self.assertNotIn("active_work_seconds", current)
|
|
self.assertIn("app-server", current["command"])
|
|
self.assertNotIn("exec", current["command"])
|
|
self.assertNotIn("--output-schema", current["command"])
|
|
self.assertNotIn("output_schema_path", current)
|
|
worker_mcp = mcp_exchange(
|
|
session,
|
|
"implementation_specialist",
|
|
caller_token=caller_token,
|
|
caller_job_id=job["job_id"],
|
|
)
|
|
self.assertEqual(worker_mcp.returncode, 0, worker_mcp.stderr)
|
|
worker_rows = [json.loads(line) for line in worker_mcp.stdout.splitlines()]
|
|
worker_tools = [
|
|
item["name"]
|
|
for row in worker_rows
|
|
if row.get("id") == 2
|
|
for item in row["result"]["tools"]
|
|
]
|
|
self.assertNotIn("agent_progress", worker_tools)
|
|
self.assertNotIn("agent_progress_history", worker_tools)
|
|
self.assertIn("agent_inspect", worker_tools)
|
|
self.assertIn("agent_steer", worker_tools)
|
|
self.assertIn("agent_result", worker_tools)
|
|
self.assertNotIn("agent_cancel", worker_tools)
|
|
self.assertNotIn("agent_spawn", worker_tools)
|
|
self.assertNotIn("agents_spawn", worker_tools)
|
|
self.assertNotIn("agent_result_accept", worker_tools)
|
|
self.assertNotIn("agent_result_reject", worker_tools)
|
|
self.assertNotIn("agent_patch_integrate", worker_tools)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["contract_transport"], "native_schema_projection")
|
|
self.assertEqual(final["result_kind"], "final")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_worker_app_server_host_is_owned_during_protocol_bootstrap(self) -> None:
|
|
with (
|
|
RuntimeSandbox() as box,
|
|
mock.patch.dict(
|
|
os.environ,
|
|
{"FAKE_CODEX_APP_SERVER_INITIALIZE_DELAY": "5"},
|
|
),
|
|
):
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Collect one bounded observation.",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
app_server_pid = current.get("app_server_pid")
|
|
if isinstance(app_server_pid, int):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not publish its app-server host during bootstrap")
|
|
self.assertEqual(current["app_server_pgid"], app_server_pid)
|
|
self.assertIsInstance(current.get("app_server_start_token"), str)
|
|
self.assertNotIn("app_server_thread_id", current)
|
|
|
|
cancelled = cancel_job(job["job_id"], session_id=session["session_id"])
|
|
self.assertEqual(cancelled["jobs"][-1]["status"], "cancelled")
|
|
self.assertFalse(process_group_alive(app_server_pid))
|
|
finally:
|
|
if load_job(job["job_id"])["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_warn_contract_does_not_enable_constrained_generation(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "warn-contract-profile"
|
|
shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "warn-contract-profile"
|
|
profile_data["agents"]["implementation_specialist"].update(
|
|
contract_enforcement="warn",
|
|
output_contract="contracts/evidence.json",
|
|
)
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Inspect one bounded invariant and return evidence without writing.",
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertNotIn("structured_output_supported", final)
|
|
self.assertNotIn("--output-schema", final["command"])
|
|
self.assertNotIn("output_schema_path", final)
|
|
self.assertEqual(final["contract_transport"], "validated_text")
|
|
event_rows = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row["message"]["params"]
|
|
for row in event_rows
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertTrue(starts)
|
|
self.assertNotIn("outputSchema", starts[0])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_strict_validated_text_contract_receives_one_same_thread_repair(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "validated-text-repair-profile"
|
|
shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "validated-text-repair-profile"
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
contract_path = profile / "contracts" / "engineering.json"
|
|
contract = {
|
|
"title": "Validated-text list result",
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"maxItems": 20,
|
|
}
|
|
contract_path.write_text(json.dumps(contract, indent=2) + "\n", encoding="utf-8")
|
|
|
|
session = create_session(profile=profile, cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task=(
|
|
"Return one bounded engineering result without writing. "
|
|
"FAKE_INVALID_FIRST_RESULT"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["contract_transport"], "validated_text")
|
|
self.assertTrue(final["contract_repair_attempted"])
|
|
self.assertTrue(final["contract_valid"])
|
|
event_rows = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row["message"]
|
|
for row in event_rows
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 2)
|
|
self.assertTrue(all("outputSchema" not in row["params"] for row in starts))
|
|
self.assertIn(
|
|
"Repair only the final JSON result",
|
|
starts[-1]["params"]["input"][0]["text"],
|
|
)
|
|
history = json.loads(
|
|
(mmo_runtime.job_dir(job["job_id"]) / "terminal-history.json").read_text()
|
|
)
|
|
self.assertEqual(len(history["turns"]), 2)
|
|
repair_messages = [
|
|
item["text"]
|
|
for item in history["turns"][-1]["items"]
|
|
if item.get("type") == "agentMessage"
|
|
]
|
|
self.assertEqual(
|
|
repair_messages[-1],
|
|
Path(final["result_path"]).read_text(encoding="utf-8"),
|
|
)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_worker_app_server_schema_trace_and_control_lifecycle(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
root_mcp = mcp_exchange(session, session["root_agent"])
|
|
root_rows = [json.loads(line) for line in root_mcp.stdout.splitlines()]
|
|
root_tools = {
|
|
item["name"]: item
|
|
for row in root_rows
|
|
if row.get("id") == 2
|
|
for item in row["result"]["tools"]
|
|
}
|
|
self.assertIn("agent_inspect", root_tools)
|
|
self.assertIn("agent_trace", root_tools)
|
|
self.assertIn("agent_steer", root_tools)
|
|
self.assertIn("agent_interrupt", root_tools)
|
|
self.assertIn("agent_pause", root_tools)
|
|
self.assertIn("agent_detach", root_tools)
|
|
self.assertIn("agent_stop", root_tools)
|
|
self.assertIn("agent_continue", root_tools)
|
|
self.assertNotIn("agent_progress", root_tools)
|
|
self.assertNotIn("agent_progress_history", root_tools)
|
|
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Collect one bounded observation from README.md and return the required "
|
|
"evidence contract. FAKE_SLEEP_SECONDS=30"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(200):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("app-server worker did not expose a controllable active turn")
|
|
|
|
self.assertEqual(current["contract_transport"], "native_schema_projection")
|
|
inspected = inspect_job(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(inspected["live"]["active_turn_id"], current["active_turn_id"])
|
|
trace = read_trace(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
limit=20,
|
|
)
|
|
self.assertTrue(trace["records"])
|
|
self.assertFalse(trace["private_reasoning_included"])
|
|
|
|
event_path = Path(current["events_path"])
|
|
append_jsonl(
|
|
event_path,
|
|
{
|
|
"recorded_at": "oversized-test",
|
|
"direction": "received",
|
|
"message": {
|
|
"jsonrpc": "2.0",
|
|
"method": "item/completed",
|
|
"params": {"tool_output": "x" * (600 * 1024)},
|
|
},
|
|
},
|
|
)
|
|
with event_path.open(encoding="utf-8") as handle:
|
|
oversized_cursor = next(
|
|
index
|
|
for index, line in enumerate(handle)
|
|
if json.loads(line).get("recorded_at") == "oversized-test"
|
|
)
|
|
oversized = read_trace(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
cursor=oversized_cursor,
|
|
limit=1,
|
|
)
|
|
self.assertEqual(oversized["next_cursor"], oversized_cursor + 1)
|
|
self.assertTrue(oversized["records"][0]["trace_record_truncated"])
|
|
self.assertLess(len(json.dumps(oversized["records"]).encode("utf-8")), 512 * 1024)
|
|
record_cursor = oversized["records"][0]["record_cursor"]
|
|
record_pages: list[str] = []
|
|
record_page_cursor = 0
|
|
record_sha256 = None
|
|
while True:
|
|
page = read_agent_trace_record(
|
|
current["agent_run_ref"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
record_cursor=record_cursor,
|
|
cursor=record_page_cursor,
|
|
max_chars=30000,
|
|
)
|
|
record_pages.append(page["content"])
|
|
record_sha256 = record_sha256 or page["filtered_sha256"]
|
|
self.assertEqual(page["filtered_sha256"], record_sha256)
|
|
if page["next_cursor"] is None:
|
|
break
|
|
record_page_cursor = page["next_cursor"]
|
|
full_record = "".join(record_pages)
|
|
self.assertEqual(len(full_record), page["total_chars"])
|
|
self.assertEqual(
|
|
json.loads(full_record)["message"]["params"]["tool_output"],
|
|
"x" * (600 * 1024),
|
|
)
|
|
|
|
malformed_raw = '{"broken":"' + "y" * (300 * 1024) + "\n"
|
|
with event_path.open("a", encoding="utf-8") as handle:
|
|
handle.write(malformed_raw)
|
|
with event_path.open(encoding="utf-8") as handle:
|
|
malformed_cursor = sum(1 for _line in handle) - 1
|
|
malformed = read_trace(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
cursor=malformed_cursor,
|
|
limit=1,
|
|
)
|
|
self.assertFalse(malformed["records"][0].get("trace_record_truncated", False))
|
|
malformed_page = read_agent_trace_record(
|
|
current["agent_run_ref"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
record_cursor=malformed_cursor,
|
|
max_chars=30000,
|
|
)
|
|
self.assertEqual(
|
|
json.loads(malformed_page["content"])["malformed_event"],
|
|
malformed_raw[:2000],
|
|
)
|
|
self.assertIsNone(malformed_page["next_cursor"])
|
|
|
|
effort = control_job(
|
|
job["job_id"],
|
|
"set_effort",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
effort="max",
|
|
)
|
|
self.assertEqual(effort["control_revision"], 1)
|
|
steered = control_job(
|
|
job["job_id"],
|
|
"steer",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=1,
|
|
input="Also state any evidence limitations explicitly.",
|
|
)
|
|
self.assertEqual(steered["control_revision"], 2)
|
|
paused_control = control_job(
|
|
job["job_id"],
|
|
"pause",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=2,
|
|
)
|
|
self.assertEqual(paused_control["control_revision"], 3)
|
|
for _ in range(200):
|
|
paused = load_job(job["job_id"])
|
|
if paused["status"] == "paused":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(paused["status"], "paused")
|
|
with self.assertRaisesRegex(RuntimeError, "stale control revision"):
|
|
control_job(
|
|
job["job_id"],
|
|
"continue",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=2,
|
|
input="This stale mutation must not run.",
|
|
)
|
|
continued = control_job(
|
|
job["job_id"],
|
|
"continue",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=3,
|
|
input=("Complete now from retained evidence. FAKE_NOTIFY_BEFORE_RESPONSE"),
|
|
)
|
|
self.assertEqual(continued["control_revision"], 4)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertTrue(final["contract_valid"])
|
|
self.assertEqual(final["result_kind"], "final")
|
|
self.assertGreaterEqual(final["usage"]["input_tokens"], 202)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_turn_worker_interrupt_remains_resumable_on_the_same_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Collect one bounded observation. FAKE_SLEEP_SECONDS=4",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("turn worker did not expose its active turn")
|
|
thread_id = current["app_server_thread_id"]
|
|
|
|
interrupted = control_job(
|
|
job["job_id"],
|
|
"interrupt",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
self.assertEqual(interrupted["control_revision"], 1)
|
|
for _ in range(300):
|
|
paused = load_job(job["job_id"])
|
|
if paused.get("status") == "paused" and not paused.get("active_turn_id"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("interrupted turn worker did not become resumably paused")
|
|
self.assertEqual(paused["app_server_thread_id"], thread_id)
|
|
|
|
continued = control_job(
|
|
job["job_id"],
|
|
"continue",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=1,
|
|
input="Complete from the retained thread context.",
|
|
)
|
|
self.assertEqual(continued["control_revision"], 2)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["app_server_thread_id"], thread_id)
|
|
self.assertTrue(final["contract_valid"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_app_server_compact_and_finalize_controls_are_delivered(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Collect one bounded observation. FAKE_SLEEP_SECONDS=1",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("app-server worker did not expose its control socket")
|
|
|
|
compacted = control_job(
|
|
job["job_id"],
|
|
"compact",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
self.assertEqual(compacted["control_revision"], 1)
|
|
finalized = control_job(
|
|
job["job_id"],
|
|
"finalize",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=1,
|
|
input="Finalize from the evidence already obtained.",
|
|
)
|
|
self.assertEqual(finalized["control_revision"], 2)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertNotIn("active_work_cap_seconds", final)
|
|
self.assertNotIn("automatic_renewal_enabled", final)
|
|
events = Path(final["events_path"]).read_text(encoding="utf-8")
|
|
self.assertIn('"method":"thread/compact/start"', events)
|
|
self.assertIn('"method":"turn/steer"', events)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_goal_finalize_interrupts_unconstrained_work_and_uses_terminal_schema(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Inspect one bounded invariant. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("goal worker did not expose its active turn")
|
|
control_socket = Path(current["control_socket_path"])
|
|
self.assertLess(len(os.fsencode(control_socket)), 104)
|
|
self.assertNotEqual(control_socket.parent, mmo_runtime.job_dir(job["job_id"]))
|
|
|
|
control_job(
|
|
job["job_id"],
|
|
"finalize",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
input="Serialize the strongest supported result from retained evidence only.",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertTrue(final["contract_valid"])
|
|
event_rows = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
sent = [
|
|
row["message"]
|
|
for row in event_rows
|
|
if row.get("direction") == "sent" and isinstance(row.get("message"), dict)
|
|
]
|
|
self.assertIn("turn/interrupt", [row.get("method") for row in sent])
|
|
starts = [row for row in sent if row.get("method") == "turn/start"]
|
|
self.assertGreaterEqual(len(starts), 2)
|
|
self.assertNotIn("outputSchema", starts[0]["params"])
|
|
self.assertIn("outputSchema", starts[-1]["params"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_session_visibility_does_not_grant_peer_result_disposition(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="route-resilience-lab", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
peer = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="go_route",
|
|
task_kind="route_probe",
|
|
task="Remain active while the peer result is reviewed. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
target = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="openrouter_route",
|
|
task_kind="route_probe",
|
|
task="Return one bounded route observation for peer review.",
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[target["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
reviewed = read_result(
|
|
target["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_job_id=peer["job_id"],
|
|
caller_agent="go_route",
|
|
)
|
|
self.assertTrue(reviewed["content"])
|
|
with self.assertRaisesRegex(PermissionError, "not a descendant"):
|
|
accept_result(
|
|
target["job_id"],
|
|
"A peer may read this session-visible result but cannot accept it.",
|
|
session_id=session["session_id"],
|
|
caller_job_id=peer["job_id"],
|
|
caller_agent="go_route",
|
|
)
|
|
accepted = accept_result(
|
|
target["job_id"],
|
|
"The root retains result-disposition authority.",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(accepted["result_state"], "accepted")
|
|
cancel_job(peer["job_id"], session_id=session["session_id"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_control_grant_does_not_bypass_lineage_cancellation_authority(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
jobs: list[str] = []
|
|
try:
|
|
controller = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Remain active while control authorization is checked. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
target = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="adversarial_reviewer",
|
|
task_kind="review",
|
|
task="Remain active as a sibling control target. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
jobs.extend([controller["job_id"], target["job_id"]])
|
|
with self.assertRaisesRegex(PermissionError, "not a descendant"):
|
|
cancel_job(
|
|
target["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_job_id=controller["job_id"],
|
|
caller_agent="implementation_specialist",
|
|
)
|
|
self.assertNotIn(
|
|
load_job(target["job_id"])["status"], mmo_runtime.TERMINAL_JOB_STATUSES
|
|
)
|
|
finally:
|
|
for job_id in jobs:
|
|
if load_job(job_id)["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job_id, session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_app_server_fork_uses_persisted_thread_and_normal_admission(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
source = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Collect one bounded source observation.",
|
|
mode="read-only",
|
|
)
|
|
source_wait = wait_for_jobs(
|
|
[source["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(source_wait["unfinished"], source_wait)
|
|
source_state = load_job(source["job_id"])
|
|
forked = fork_job(
|
|
source["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
expected_revision=0,
|
|
task_kind="evidence_collection",
|
|
task="Independently challenge the retained source context.",
|
|
)
|
|
fork_wait = wait_for_jobs(
|
|
[forked["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(fork_wait["unfinished"], fork_wait)
|
|
fork_state = load_job(forked["job_id"])
|
|
self.assertEqual(fork_state["status"], "completed")
|
|
self.assertEqual(fork_state["fork_source_job_id"], source["job_id"])
|
|
self.assertEqual(fork_state["fork_thread_id"], source_state["app_server_thread_id"])
|
|
self.assertNotEqual(
|
|
fork_state["app_server_thread_id"], source_state["app_server_thread_id"]
|
|
)
|
|
self.assertEqual(load_job(source["job_id"])["control_revision"], 1)
|
|
self.assertIn(
|
|
'"method":"thread/fork"',
|
|
Path(fork_state["events_path"]).read_text(encoding="utf-8"),
|
|
)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_pending_app_server_input_retains_thread_and_requires_typed_response(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Request one synthetic clarification, then return a bounded evidence "
|
|
"record. FAKE_REQUEST_USER_INPUT"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
first = inspect_job(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
pending = (first.get("live") or {}).get("pending_requests", [])
|
|
if pending:
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose its pending app-server request")
|
|
self.assertNotIn("active_work_seconds", first["live"])
|
|
thread_id = first["live"]["thread_id"]
|
|
request_id = pending[0]["id"]
|
|
time.sleep(0.5)
|
|
second = inspect_job(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(second["live"]["thread_id"], thread_id)
|
|
self.assertEqual(second["live"]["pending_requests"][0]["id"], request_id)
|
|
self.assertEqual(load_job(job["job_id"])["status"], "waiting")
|
|
with self.assertRaisesRegex(RuntimeError, "invalid fields"):
|
|
control_job(
|
|
job["job_id"],
|
|
"respond",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
request_id=request_id,
|
|
response={"decision": "decline"},
|
|
)
|
|
response = control_job(
|
|
job["job_id"],
|
|
"respond",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=1,
|
|
request_id=request_id,
|
|
response={"answers": {}},
|
|
)
|
|
self.assertEqual(response["control_revision"], 2)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
self.assertEqual(load_job(job["job_id"])["status"], "completed")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_app_server_approval_policy_is_enforced_and_controller_answerable(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
denied = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Exercise automatic denial and still return evidence. "
|
|
"FAKE_REQUEST_COMMAND_APPROVAL"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[denied["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(denied["job_id"])
|
|
self.assertEqual(final["approval_policy"], "never")
|
|
sent = [
|
|
row["message"]
|
|
for row in map(
|
|
json.loads,
|
|
Path(final["events_path"]).read_text(encoding="utf-8").splitlines(),
|
|
)
|
|
if row.get("direction") == "sent"
|
|
and str(row.get("message", {}).get("id", "")).startswith("fake-approval-")
|
|
]
|
|
self.assertEqual(sent[0]["result"], {"decision": "decline"})
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
with RuntimeSandbox() as box:
|
|
destination = clone_profile("incident-hypothesis-triage", "approval-controller-test")
|
|
profile = read_toml(destination / "profile.toml")
|
|
profile["agents"]["evidence_runner"]["approval_policy"] = "on-request"
|
|
(destination / "profile.toml").write_text(toml_dumps(profile), encoding="utf-8")
|
|
session = create_session(profile="approval-controller-test", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Expose one approval request, then return evidence. "
|
|
"FAKE_REQUEST_COMMAND_APPROVAL"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
observed = inspect_job(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
pending = (observed.get("live") or {}).get("pending_requests", [])
|
|
if pending:
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose its pending approval request")
|
|
self.assertEqual(pending[0]["method"], "item/commandExecution/requestApproval")
|
|
answered = control_job(
|
|
job["job_id"],
|
|
"respond",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
request_id=pending[0]["id"],
|
|
response={"decision": "decline"},
|
|
)
|
|
self.assertEqual(answered["control_revision"], 1)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed")
|
|
self.assertEqual(final["approval_policy"], "on-request")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_worker_transport_recovers_same_persisted_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Recover once from the synthetic app-server transport failure and return "
|
|
"bounded evidence. FAKE_APP_SERVER_CRASH_ONCE"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if isinstance(current.get("app_server_thread_id"), str):
|
|
thread_id = current["app_server_thread_id"]
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not persist its app-server thread")
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=15,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["app_server_thread_id"], thread_id)
|
|
self.assertEqual(final["recovery_attempts"], 1)
|
|
self.assertIsNone(final.get("recovery_error"))
|
|
self.assertNotIn("active_work_seconds", final)
|
|
history = json.loads(
|
|
(mmo_runtime.job_dir(job["job_id"]) / "terminal-history.json").read_text()
|
|
)
|
|
self.assertEqual(history["thread_id"], thread_id)
|
|
self.assertTrue(history["turns"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_detached_worker_recovers_transport_and_keeps_working(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Recover after this worker is detached and return bounded evidence. "
|
|
"FAKE_SLEEP_SECONDS=1 FAKE_APP_SERVER_CRASH_ONCE"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not expose its active turn before detachment")
|
|
thread_id = current["app_server_thread_id"]
|
|
detached = control_job(
|
|
job["job_id"],
|
|
"detach",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
self.assertEqual(detached["job"]["status"], "detached")
|
|
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=15,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["app_server_thread_id"], thread_id)
|
|
self.assertEqual(final["recovery_attempts"], 1)
|
|
self.assertTrue(final["contract_valid"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_worker_recovery_consumes_terminal_turn_persisted_before_transport_loss(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Return this result once, then lose only the completion notification. "
|
|
"FAKE_APP_SERVER_CRASH_AFTER_PERSIST_ONCE"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=15,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["recovery_attempts"], 1)
|
|
self.assertNotIn("app_server_rollout_path", final)
|
|
events = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 1, starts)
|
|
history = json.loads(
|
|
(mmo_runtime.job_dir(job["job_id"]) / "terminal-history.json").read_text()
|
|
)
|
|
self.assertEqual(len(history["turns"]), 1)
|
|
self.assertEqual(final["last_turn_id"], history["turns"][0]["id"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_root_recovery_consumes_terminal_turn_persisted_before_transport_loss(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt=(
|
|
"Return this root result once, then lose only the completion notification. "
|
|
"FAKE_APP_SERVER_CRASH_AFTER_PERSIST_ONCE"
|
|
),
|
|
wall_timeout_seconds=15,
|
|
)
|
|
self.assertEqual(result["status"], "completed", result)
|
|
session = load_session(result["session"]["session_id"])
|
|
self.assertNotIn("root_rollout_path", session)
|
|
events = [
|
|
json.loads(line)
|
|
for line in Path(result["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 1, starts)
|
|
history = json.loads(
|
|
(
|
|
box.state / "sessions" / session["session_id"] / "root-terminal-history.json"
|
|
).read_text()
|
|
)
|
|
self.assertEqual(len(history["turns"]), 1)
|
|
self.assertEqual(session["root_last_turn_id"], history["turns"][0]["id"])
|
|
|
|
def test_root_completion_without_a_readable_result_fails_with_retained_evidence(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Complete without a terminal message. FAKE_EMPTY_RESULT",
|
|
wall_timeout_seconds=15,
|
|
)
|
|
self.assertEqual(result["status"], "failed", result)
|
|
session = load_session(result["session"]["session_id"])
|
|
self.assertEqual(session["result_kind"], "partial")
|
|
self.assertIn("without a readable terminal result", session["error"])
|
|
partial = Path(session["result_path"])
|
|
self.assertTrue(partial.is_file())
|
|
self.assertIn(
|
|
"root completed without a readable terminal result",
|
|
partial.read_text(encoding="utf-8"),
|
|
)
|
|
|
|
def test_worker_host_loss_retains_partial_and_relaunches_with_control_prompt(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Retain this delegated context across one synthetic host loss and return "
|
|
"bounded evidence. FAKE_SLEEP_SECONDS=1.5"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and isinstance(current.get("runner_pid"), int):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not begin an app-server turn")
|
|
thread_id = current["app_server_thread_id"]
|
|
original_started_at = current["started_at"]
|
|
old_app_server_pid = current["app_server_pid"]
|
|
old_caller_hash = current["mcp_caller_token_hash"]
|
|
self.assertFalse((mmo_runtime.job_dir(job["job_id"]) / "caller-token").exists())
|
|
os.killpg(current["runner_pid"], signal.SIGKILL)
|
|
for _ in range(300):
|
|
suspended = load_job(job["job_id"])
|
|
if suspended["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(suspended["status"], "suspended")
|
|
self.assertFalse(process_group_alive(old_app_server_pid))
|
|
partial_path = Path(suspended["partial_result_path"])
|
|
self.assertTrue(partial_path.is_file())
|
|
self.assertIn("persisted app-server thread", partial_path.read_text())
|
|
|
|
continued = control_job(
|
|
job["job_id"],
|
|
"continue",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
input="RECOVERY_MARKER Complete from the retained evidence now.",
|
|
)
|
|
self.assertTrue(continued["result"]["recovery_requested"])
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=15,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(final["app_server_thread_id"], thread_id)
|
|
self.assertEqual(final["started_at"], original_started_at)
|
|
self.assertEqual(final["last_control_status"], "applied")
|
|
self.assertEqual(
|
|
Path(final["control_socket_path"]),
|
|
app_server_socket_path(f"control:job:{job['job_id']}"),
|
|
)
|
|
self.assertNotEqual(final["mcp_caller_token_hash"], old_caller_hash)
|
|
self.assertFalse((mmo_runtime.job_dir(job["job_id"]) / "caller-token").exists())
|
|
self.assertIsNone(final.get("recovery_prompt"))
|
|
self.assertIn(
|
|
"RECOVERY_MARKER",
|
|
Path(final["events_path"]).read_text(encoding="utf-8"),
|
|
)
|
|
self.assertTrue(partial_path.is_file())
|
|
lifecycle_events = [
|
|
json.loads(line)
|
|
for line in Path(session["audit_path"]).read_text(encoding="utf-8").splitlines()
|
|
if job["job_id"] in line
|
|
and any(name in line for name in ('"agent_started"', '"agent_resumed"'))
|
|
]
|
|
self.assertEqual(
|
|
[row["event"] for row in lifecycle_events],
|
|
["agent_started", "agent_resumed"],
|
|
)
|
|
runner_log = mmo_runtime.job_dir(job["job_id"]) / "runner.log"
|
|
self.assertNotIn("Traceback", runner_log.read_text(encoding="utf-8"))
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_budget_limited_goal_publishes_valid_terminal_turn_with_warning(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="causal_challenger",
|
|
task_kind="causal_analysis",
|
|
task=(
|
|
"Return strict bounded hypothesis evidence. "
|
|
"FAKE_SLEEP_SECONDS=0.2 FAKE_GOAL_BUDGET_LIMIT_BEFORE_TURN"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed_with_warnings", final)
|
|
self.assertEqual(final["goal_status"], "budgetLimited")
|
|
self.assertTrue(final["contract_valid"])
|
|
self.assertEqual(final["result_kind"], "final")
|
|
self.assertIn("fully validated strict result", final["warning"])
|
|
self.assertTrue(Path(final["result_path"]).is_file())
|
|
self.assertTrue(Path(final["structured_result_path"]).is_file())
|
|
event_text = Path(final["events_path"]).read_text(encoding="utf-8")
|
|
events = [json.loads(line) for line in event_text.splitlines()]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 1)
|
|
self.assertNotIn("Repair only the final JSON result", event_text)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_budget_limited_goal_recovers_persisted_turn_after_transport_loss(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="causal_challenger",
|
|
task_kind="causal_analysis",
|
|
task=(
|
|
"Persist the strict terminal turn before one synthetic host loss. "
|
|
"FAKE_SLEEP_SECONDS=0.2 FAKE_GOAL_BUDGET_LIMIT_BEFORE_TURN "
|
|
"FAKE_APP_SERVER_CRASH_AFTER_PERSIST_ONCE"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed_with_warnings", final)
|
|
self.assertEqual(final["goal_status"], "budgetLimited")
|
|
self.assertTrue(final["contract_valid"])
|
|
self.assertGreaterEqual(final["recovery_attempts"], 1)
|
|
events = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 1, starts)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_suspended_budget_result_can_finalize_on_same_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="causal_challenger",
|
|
task_kind="causal_analysis",
|
|
task=(
|
|
"Retain evidence, then require evidence-only serialization. "
|
|
"FAKE_SLEEP_SECONDS=0.2 FAKE_INVALID_FIRST_RESULT "
|
|
"FAKE_GOAL_BUDGET_LIMIT_BEFORE_TURN"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(500):
|
|
suspended = load_job(job["job_id"])
|
|
if suspended["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(suspended["status"], "suspended", suspended)
|
|
self.assertEqual(suspended["goal_status"], "budgetLimited")
|
|
self.assertEqual(suspended["result_kind"], "partial")
|
|
thread_id = suspended["app_server_thread_id"]
|
|
|
|
requested = control_job(
|
|
job["job_id"],
|
|
"finalize",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
input="Serialize only the evidence already retained in this thread.",
|
|
)
|
|
self.assertEqual(requested["result"]["action"], "finalize")
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed_with_warnings", final)
|
|
self.assertEqual(final["app_server_thread_id"], thread_id)
|
|
self.assertEqual(final["goal_status"], "budgetLimited")
|
|
self.assertEqual(final["last_control_status"], "applied")
|
|
self.assertTrue(final["contract_valid"])
|
|
self.assertTrue(Path(final["partial_result_path"]).is_file())
|
|
self.assertTrue(Path(final["structured_result_path"]).is_file())
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_recovery_control_acknowledges_only_its_exact_pending_revision(self) -> None:
|
|
with RuntimeSandbox():
|
|
directory = mmo_runtime.jobs_root() / "audit-recovery-control"
|
|
directory.mkdir(mode=0o700)
|
|
publish_job_record(
|
|
directory,
|
|
{
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": directory.name,
|
|
"control_revision": 2,
|
|
"last_control_action": "continue",
|
|
"last_control_status": "pending",
|
|
},
|
|
)
|
|
recovery_state = {
|
|
"recovery_control_revision": 1,
|
|
"recovery_action": "continue",
|
|
}
|
|
self.assertFalse(
|
|
worker_runner._settle_recovery_control(
|
|
directory,
|
|
recovery_state,
|
|
status="applied",
|
|
)
|
|
)
|
|
self.assertEqual(read_job_record(directory)["last_control_status"], "pending")
|
|
|
|
recovery_state["recovery_control_revision"] = 2
|
|
self.assertTrue(
|
|
worker_runner._settle_recovery_control(
|
|
directory,
|
|
recovery_state,
|
|
status="applied",
|
|
)
|
|
)
|
|
self.assertEqual(read_job_record(directory)["last_control_status"], "applied")
|
|
|
|
def test_retiring_control_server_cannot_unlink_replacement_generation(self) -> None:
|
|
with RuntimeSandbox():
|
|
directory = mmo_runtime.jobs_root() / "audit-control-generation"
|
|
directory.mkdir(mode=0o700)
|
|
socket_path = app_server_socket_path(f"control:job:{directory.name}")
|
|
publish_job_record(
|
|
directory,
|
|
{
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": directory.name,
|
|
"control_socket_path": str(socket_path),
|
|
"control_socket_ready": False,
|
|
},
|
|
)
|
|
state: dict[str, Any] = {}
|
|
state_lock = threading.RLock()
|
|
errors: list[BaseException] = []
|
|
|
|
def start_server(stop: threading.Event, ready: threading.Event) -> threading.Thread:
|
|
def serve() -> None:
|
|
try:
|
|
worker_runner._serve_control(
|
|
directory,
|
|
state,
|
|
state_lock,
|
|
stop,
|
|
ready,
|
|
)
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
ready.set()
|
|
|
|
thread = threading.Thread(target=serve, daemon=True)
|
|
thread.start()
|
|
self.assertTrue(ready.wait(timeout=5))
|
|
self.assertFalse(errors, errors)
|
|
return thread
|
|
|
|
first_stop = threading.Event()
|
|
first = start_server(first_stop, threading.Event())
|
|
self.assertTrue(socket_path.is_socket())
|
|
|
|
socket_path.unlink()
|
|
current = read_job_record(directory)
|
|
current["control_socket_ready"] = False
|
|
publish_job_record(directory, current)
|
|
|
|
second_stop = threading.Event()
|
|
second = start_server(second_stop, threading.Event())
|
|
replacement_identity = socket_path.stat(follow_symlinks=False).st_ino
|
|
first_stop.set()
|
|
first.join(timeout=3)
|
|
self.assertFalse(first.is_alive())
|
|
self.assertTrue(socket_path.is_socket())
|
|
self.assertEqual(
|
|
socket_path.stat(follow_symlinks=False).st_ino,
|
|
replacement_identity,
|
|
)
|
|
self.assertTrue(read_job_record(directory)["control_socket_ready"])
|
|
|
|
second_stop.set()
|
|
second.join(timeout=3)
|
|
self.assertFalse(second.is_alive())
|
|
self.assertFalse(socket_path.exists())
|
|
self.assertFalse(read_job_record(directory)["control_socket_ready"])
|
|
self.assertFalse(errors, errors)
|
|
|
|
def test_full_stop_terminalizes_a_suspended_worker_without_discarding_evidence(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Retain partial evidence across a host loss. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("active_turn_id") and isinstance(current.get("runner_pid"), int):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not begin an app-server turn")
|
|
os.killpg(current["runner_pid"], signal.SIGKILL)
|
|
for _ in range(300):
|
|
suspended = load_job(job["job_id"])
|
|
if suspended["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(suspended["status"], "suspended")
|
|
partial_path = Path(suspended["partial_result_path"])
|
|
stopped = control_job(
|
|
job["job_id"],
|
|
"stop",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
self.assertEqual(stopped["job"]["status"], "stopped")
|
|
self.assertEqual(load_job(job["job_id"])["status"], "stopped")
|
|
self.assertTrue(partial_path.is_file())
|
|
finally:
|
|
if load_job(job["job_id"])["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_mutating_controls_are_serialized_through_delivery(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Remain active for serialized controls. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
first_entered = threading.Event()
|
|
release_first = threading.Event()
|
|
second_entered = threading.Event()
|
|
call_count = 0
|
|
call_lock = threading.Lock()
|
|
|
|
def controlled_delivery(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
|
nonlocal call_count
|
|
with call_lock:
|
|
call_count += 1
|
|
ordinal = call_count
|
|
if ordinal == 1:
|
|
first_entered.set()
|
|
self.assertTrue(release_first.wait(5))
|
|
else:
|
|
second_entered.set()
|
|
return {"ok": True, "result": {}}
|
|
|
|
errors: list[BaseException] = []
|
|
|
|
def mutate(revision: int) -> None:
|
|
try:
|
|
control_job(
|
|
job["job_id"],
|
|
"compact",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=revision,
|
|
)
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
with mock.patch.object(
|
|
mmo_runtime, "send_control_request", side_effect=controlled_delivery
|
|
):
|
|
first = threading.Thread(target=mutate, args=(0,))
|
|
second = threading.Thread(target=mutate, args=(1,))
|
|
first.start()
|
|
self.assertTrue(first_entered.wait(5))
|
|
second.start()
|
|
self.assertFalse(second_entered.wait(0.25))
|
|
release_first.set()
|
|
first.join(timeout=5)
|
|
second.join(timeout=5)
|
|
self.assertFalse(errors, errors)
|
|
self.assertTrue(second_entered.is_set())
|
|
self.assertEqual(load_job(job["job_id"])["control_revision"], 2)
|
|
finally:
|
|
if load_job(job["job_id"])["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_ambiguous_control_reply_is_not_recorded_as_definite_failure(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Remain active for one ambiguous control. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
try:
|
|
for _ in range(300):
|
|
current = load_job(job["job_id"])
|
|
if current.get("control_socket_ready"):
|
|
break
|
|
time.sleep(0.02)
|
|
with mock.patch.object(
|
|
mmo_runtime,
|
|
"send_control_request",
|
|
side_effect=ControlDeliveryUnknown("reply lost"),
|
|
):
|
|
with self.assertRaises(ControlDeliveryUnknown):
|
|
control_job(
|
|
job["job_id"],
|
|
"compact",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
uncertain = load_job(job["job_id"])
|
|
self.assertEqual(uncertain["last_control_status"], "delivery_unknown")
|
|
self.assertEqual(uncertain["control_revision"], 1)
|
|
finally:
|
|
if load_job(job["job_id"])["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job["job_id"], session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_suspended_worker_continuation_is_readmitted_before_relaunch(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
jobs: list[str] = []
|
|
try:
|
|
suspended_job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task="Retain evidence across a host loss. FAKE_SLEEP_SECONDS=5",
|
|
mode="read-only",
|
|
)
|
|
jobs.append(suspended_job["job_id"])
|
|
for _ in range(300):
|
|
current = load_job(suspended_job["job_id"])
|
|
if current.get("active_turn_id") and isinstance(current.get("runner_pid"), int):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("worker did not begin an app-server turn")
|
|
os.killpg(current["runner_pid"], signal.SIGKILL)
|
|
for _ in range(300):
|
|
current = load_job(suspended_job["job_id"])
|
|
if current["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(current["status"], "suspended")
|
|
|
|
active = spawn_jobs(
|
|
[
|
|
{
|
|
"agent": "evidence_runner",
|
|
"task_kind": "evidence_collection",
|
|
"task": "Collect bounded evidence. FAKE_SLEEP_SECONDS=5",
|
|
},
|
|
{
|
|
"agent": "causal_challenger",
|
|
"task_kind": "causal_analysis",
|
|
"task": "Challenge one bounded hypothesis. FAKE_SLEEP_SECONDS=5",
|
|
},
|
|
],
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
)
|
|
jobs.extend(item["job_id"] for item in active["accepted"])
|
|
with self.assertRaisesRegex(RuntimeError, "active-agent limit"):
|
|
control_job(
|
|
suspended_job["job_id"],
|
|
"continue",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
expected_revision=0,
|
|
input="Resume only if scheduler capacity is available.",
|
|
)
|
|
unchanged = load_job(suspended_job["job_id"])
|
|
self.assertEqual(unchanged["status"], "suspended")
|
|
self.assertEqual(unchanged["control_revision"], 0)
|
|
finally:
|
|
for job_id in jobs:
|
|
if load_job(job_id)["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job_id, session_id=session["session_id"])
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_suspended_writer_continuation_checks_other_sessions_for_scope_conflicts(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
first_session = create_session(
|
|
profile="access-efficient-escalation-lab", cwd=box.workspace
|
|
)
|
|
second_session = create_session(
|
|
profile="competing-implementations-lab", cwd=box.workspace
|
|
)
|
|
mark_session_running(first_session["session_id"], os.getpid())
|
|
mark_session_running(second_session["session_id"], os.getpid())
|
|
jobs: list[tuple[str, str]] = []
|
|
try:
|
|
suspended_job = spawn_job(
|
|
session_id=first_session["session_id"],
|
|
caller_agent=first_session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task="Hold an isolated writer lease. FAKE_SLEEP_SECONDS=5",
|
|
mode="workspace-write",
|
|
write_scope_values=["shared.py"],
|
|
)
|
|
jobs.append((suspended_job["job_id"], first_session["session_id"]))
|
|
for _ in range(300):
|
|
current = load_job(suspended_job["job_id"])
|
|
if current.get("active_turn_id") and isinstance(current.get("runner_pid"), int):
|
|
break
|
|
time.sleep(0.02)
|
|
os.killpg(current["runner_pid"], signal.SIGKILL)
|
|
for _ in range(300):
|
|
current = load_job(suspended_job["job_id"])
|
|
if current["status"] == "suspended":
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(current["status"], "suspended")
|
|
|
|
active_writer = spawn_job(
|
|
session_id=second_session["session_id"],
|
|
caller_agent=second_session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="deepseek_candidate",
|
|
task_kind="implement",
|
|
task="Hold the overlapping cross-session scope. FAKE_SLEEP_SECONDS=5",
|
|
mode="workspace-write",
|
|
write_scope_values=["shared.py"],
|
|
)
|
|
jobs.append((active_writer["job_id"], second_session["session_id"]))
|
|
for _ in range(300):
|
|
active = load_job(active_writer["job_id"])
|
|
if active.get("active_turn_id"):
|
|
break
|
|
time.sleep(0.02)
|
|
with self.assertRaisesRegex(AdmissionError, "write scope conflicts"):
|
|
control_job(
|
|
suspended_job["job_id"],
|
|
"continue",
|
|
session_id=first_session["session_id"],
|
|
caller_agent=first_session["root_agent"],
|
|
expected_revision=0,
|
|
)
|
|
unchanged = load_job(suspended_job["job_id"])
|
|
self.assertEqual(unchanged["status"], "suspended")
|
|
self.assertEqual(unchanged["control_revision"], 0)
|
|
finally:
|
|
for job_id, session_id in jobs:
|
|
if load_job(job_id)["status"] not in mmo_runtime.TERMINAL_JOB_STATUSES:
|
|
cancel_job(job_id, session_id=session_id)
|
|
finish_session(first_session["session_id"], exit_code=0)
|
|
finish_session(second_session["session_id"], exit_code=0)
|
|
|
|
def test_graceful_stop_serializes_lifecycle_and_retains_root_evidence(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
update_session(
|
|
session["session_id"],
|
|
**root_thread_binding("00000000-0000-0000-0000-000000000010"),
|
|
)
|
|
directory = mmo_runtime.session_dir(session["session_id"])
|
|
append_jsonl(
|
|
directory / "root-events.jsonl",
|
|
{
|
|
"message": {
|
|
"params": {
|
|
"item": {
|
|
"type": "agentMessage",
|
|
"text": "durable root evidence before stop",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
)
|
|
result: list[dict[str, Any]] = []
|
|
failures: list[BaseException] = []
|
|
|
|
def stop() -> None:
|
|
try:
|
|
result.append(stop_session(session["session_id"], grace_seconds=0))
|
|
except BaseException as exc:
|
|
failures.append(exc)
|
|
|
|
with mmo_runtime.file_lock(mmo_runtime.session_lifecycle_lock_path(directory)):
|
|
thread = threading.Thread(target=stop)
|
|
thread.start()
|
|
time.sleep(0.2)
|
|
self.assertTrue(thread.is_alive())
|
|
self.assertEqual(load_session(session["session_id"])["status"], "running")
|
|
thread.join(timeout=5)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertFalse(failures, failures)
|
|
self.assertEqual(result[0]["session"]["status"], "stopped")
|
|
partial = directory / "root-partial-result.md"
|
|
self.assertTrue(partial.is_file())
|
|
self.assertIn("durable root evidence before stop", partial.read_text(encoding="utf-8"))
|
|
|
|
def test_root_wall_limit_detaches_partial_and_resumes_same_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Preserve this root task across detachment. FAKE_SLEEP_SECONDS=2",
|
|
wall_timeout_seconds=1,
|
|
)
|
|
self.assertEqual(result["status"], "detached", result)
|
|
session_id = result["session"]["session_id"]
|
|
detached = load_session(session_id)
|
|
thread_id = detached["root_thread_id"]
|
|
self.assertEqual(detached["status"], "detached")
|
|
self.assertEqual(detached["session_kind"], "noninteractive")
|
|
self.assertEqual(detached["root_execution_host"], "app_server")
|
|
self.assertEqual(detached["root_execution_mode"], "goal")
|
|
self.assertGreater(detached["root_goal_token_budget"], 0)
|
|
self.assertNotIn("root_execution_policy_enforced", detached)
|
|
partial = box.state / "sessions" / session_id / "root-partial-result.md"
|
|
self.assertTrue(partial.is_file())
|
|
self.assertIn("external harness wall expired", partial.read_text(encoding="utf-8"))
|
|
self.assertEqual(resume_interactive(session_id), 0)
|
|
deadline = time.monotonic() + 10
|
|
while load_session(session_id)["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
resumed = load_session(session_id)
|
|
self.assertEqual(resumed["root_thread_id"], thread_id)
|
|
self.assertEqual(resumed["status"], "completed")
|
|
self.assertEqual(resumed["root_execution_host"], "app_server")
|
|
|
|
def test_root_controller_crash_resume_reuses_the_live_app_server_and_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
results: list[dict[str, Any]] = []
|
|
failures: list[BaseException] = []
|
|
|
|
def run() -> None:
|
|
try:
|
|
results.append(
|
|
run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Survive a controller crash. FAKE_SLEEP_SECONDS=3",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(exc)
|
|
|
|
thread = threading.Thread(target=run)
|
|
thread.start()
|
|
for _ in range(500):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and isinstance(current.get("active_root_turn_id"), str)
|
|
and isinstance(current.get("root_app_server_pid"), int)
|
|
and current.get("root_control_socket_ready")
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("root did not publish its controller and app-server identities")
|
|
|
|
root_pid = current["root_pid"]
|
|
original_started_at = current["started_at"]
|
|
app_server_pid = current["root_app_server_pid"]
|
|
app_server_start_token = current["root_app_server_start_token"]
|
|
thread_id = current["root_thread_id"]
|
|
os.killpg(current["root_pgid"], signal.SIGKILL)
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertFalse(failures, failures)
|
|
self.assertEqual(results[0]["status"], "suspended")
|
|
suspended = load_session(current["session_id"])
|
|
self.assertEqual(suspended["root_thread_id"], thread_id)
|
|
self.assertEqual(suspended["root_app_server_pid"], app_server_pid)
|
|
self.assertTrue(process_matches(app_server_pid, app_server_start_token))
|
|
mmo_runtime._reap_tracked_runner(root_pid)
|
|
|
|
self.assertEqual(resume_interactive(current["session_id"]), 0)
|
|
resumed = load_session(current["session_id"])
|
|
resumed_root_pid = resumed.get("root_pid")
|
|
resumed_root_start_token = resumed.get("root_start_token")
|
|
self.assertEqual(resumed["root_thread_id"], thread_id)
|
|
if resumed["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
self.assertEqual(resumed["root_app_server_pid"], app_server_pid)
|
|
self.assertTrue(process_matches(app_server_pid, app_server_start_token))
|
|
deadline = time.monotonic() + 15
|
|
while load_session(current["session_id"])["status"] not in (
|
|
mmo_runtime.TERMINAL_SESSION_STATUSES
|
|
):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(current["session_id"])
|
|
self.assertEqual(completed["status"], "completed")
|
|
self.assertEqual(completed["root_thread_id"], thread_id)
|
|
self.assertEqual(completed["started_at"], original_started_at)
|
|
run_events = [
|
|
json.loads(line)["event"]
|
|
for line in Path(completed["audit_path"]).read_text(encoding="utf-8").splitlines()
|
|
if any(name in line for name in ('"run_started"', '"run_resumed"'))
|
|
]
|
|
self.assertEqual(run_events, ["run_started", "run_resumed"])
|
|
runner_deadline = time.monotonic() + 5
|
|
while process_matches(resumed_root_pid, resumed_root_start_token):
|
|
self.assertLess(time.monotonic(), runner_deadline)
|
|
time.sleep(0.05)
|
|
mmo_runtime._reap_tracked_runner(resumed_root_pid)
|
|
|
|
def test_turn_root_interrupt_remains_resumable_on_the_same_thread(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "turn-root-profile"
|
|
shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "turn-root-profile"
|
|
root_agent = profile_data["agents"]["integrator"]
|
|
root_agent["execution_mode"] = "turn"
|
|
root_agent.pop("goal_token_budget")
|
|
root_agent.pop("max_goal_token_budget")
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
|
|
results: list[dict[str, Any]] = []
|
|
failures: list[BaseException] = []
|
|
|
|
def run() -> None:
|
|
try:
|
|
results.append(
|
|
run_root_exec(
|
|
profile=profile,
|
|
cwd=box.workspace,
|
|
prompt="Retain this root turn. FAKE_SLEEP_SECONDS=4",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(exc)
|
|
|
|
thread = threading.Thread(target=run)
|
|
thread.start()
|
|
session_id: str | None = None
|
|
try:
|
|
for _ in range(500):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and isinstance(current.get("active_root_turn_id"), str)
|
|
and current.get("root_control_socket_ready")
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("turn root did not expose its active turn")
|
|
active_session_id = str(current["session_id"])
|
|
session_id = active_session_id
|
|
thread_id = current["root_thread_id"]
|
|
socket_path = Path(current["root_control_socket"])
|
|
inspection = mmo_runtime.send_control_request(socket_path, {"action": "inspect"})[
|
|
"result"
|
|
]
|
|
interrupted = mmo_runtime.send_control_request(
|
|
socket_path,
|
|
{
|
|
"action": "interrupt",
|
|
"expected_revision": inspection["control_revision"],
|
|
},
|
|
)["result"]
|
|
self.assertEqual(interrupted, {})
|
|
for _ in range(300):
|
|
paused = load_session(active_session_id)
|
|
if paused.get("status") == "paused" and not paused.get("active_root_turn_id"):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("interrupted turn root did not become resumably paused")
|
|
self.assertEqual(paused["root_thread_id"], thread_id)
|
|
self.assertTrue(thread.is_alive())
|
|
|
|
continued = mmo_runtime.send_control_request(
|
|
socket_path,
|
|
{
|
|
"action": "continue",
|
|
"expected_revision": inspection["control_revision"] + 1,
|
|
"input": "Complete from the retained thread context.",
|
|
},
|
|
)["result"]
|
|
self.assertIsInstance(continued["turn_id"], str)
|
|
thread.join(timeout=15)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertFalse(failures, failures)
|
|
self.assertEqual(results[0]["status"], "completed")
|
|
self.assertEqual(results[0]["session"]["root_thread_id"], thread_id)
|
|
finally:
|
|
if thread.is_alive() and session_id is not None:
|
|
cancel_session(session_id)
|
|
thread.join(timeout=10)
|
|
|
|
def test_detached_turn_root_recovers_transport_and_keeps_working(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
profile = box.root / "detached-turn-root-profile"
|
|
shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
|
|
profile_path = profile / "profile.toml"
|
|
profile_data = read_toml(profile_path)
|
|
profile_data["id"] = "detached-turn-root-profile"
|
|
root_agent = profile_data["agents"]["integrator"]
|
|
root_agent["execution_mode"] = "turn"
|
|
root_agent.pop("goal_token_budget")
|
|
root_agent.pop("max_goal_token_budget")
|
|
profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
|
|
|
|
result = run_root_exec(
|
|
profile=profile,
|
|
cwd=box.workspace,
|
|
prompt=(
|
|
"Recover this detached root after transport loss. "
|
|
"FAKE_SLEEP_SECONDS=1 FAKE_APP_SERVER_CRASH_ONCE"
|
|
),
|
|
wall_timeout_seconds=1,
|
|
)
|
|
self.assertEqual(result["status"], "detached", result)
|
|
session_id = result["session"]["session_id"]
|
|
thread_id = result["session"]["root_thread_id"]
|
|
deadline = time.monotonic() + 15
|
|
while load_session(session_id)["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(session_id)
|
|
self.assertEqual(completed["status"], "completed", completed)
|
|
self.assertEqual(completed["root_thread_id"], thread_id)
|
|
self.assertEqual(completed["root_recovery_attempts"], 1)
|
|
|
|
def test_root_completion_defers_until_recoverable_worker_finishes(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
root_results: list[dict[str, Any]] = []
|
|
root_failures: list[BaseException] = []
|
|
|
|
def run() -> None:
|
|
try:
|
|
root_results.append(
|
|
run_root_exec(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
prompt="Complete the root path early. FAKE_SLEEP_SECONDS=1",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
root_failures.append(exc)
|
|
|
|
thread = threading.Thread(target=run)
|
|
thread.start()
|
|
for _ in range(500):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and isinstance(current.get("active_root_turn_id"), str)
|
|
and current.get("root_control_socket_ready")
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("adaptive root did not start")
|
|
root_pid = current["root_pid"]
|
|
root_start_token = current["root_start_token"]
|
|
|
|
worker = spawn_job(
|
|
session_id=current["session_id"],
|
|
caller_agent=current["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task="Retain a recoverable worker until its evidence is ready. FAKE_SLEEP_SECONDS=3",
|
|
mode="read-only",
|
|
)
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertFalse(root_failures, root_failures)
|
|
self.assertEqual(root_results[0]["status"], "detached")
|
|
deferred = load_session(current["session_id"])
|
|
self.assertTrue(deferred["root_completion_deferred"])
|
|
self.assertIn(worker["job_id"], deferred["root_completion_deferred_jobs"])
|
|
self.assertEqual(deferred["status"], "detached")
|
|
|
|
waited = wait_for_jobs(
|
|
[worker["job_id"]],
|
|
session_id=current["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
deadline = time.monotonic() + 10
|
|
while load_session(current["session_id"])["status"] not in (
|
|
mmo_runtime.TERMINAL_SESSION_STATUSES
|
|
):
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(current["session_id"])
|
|
self.assertEqual(completed["status"], "completed")
|
|
self.assertEqual(load_job(worker["job_id"])["status"], "completed")
|
|
self.assertFalse(completed["root_completion_deferred"])
|
|
self.assertEqual(completed["root_completion_deferred_jobs"], [])
|
|
audit = [
|
|
json.loads(line)
|
|
for line in Path(completed["audit_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
self.assertEqual(
|
|
sum(row.get("event") == "root_completion_deferred" for row in audit),
|
|
1,
|
|
)
|
|
runner_deadline = time.monotonic() + 5
|
|
while process_matches(root_pid, root_start_token):
|
|
self.assertLess(time.monotonic(), runner_deadline)
|
|
time.sleep(0.05)
|
|
mmo_runtime._reap_tracked_runner(root_pid)
|
|
|
|
def test_detach_retains_root_host_and_full_stop_retires_it(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
with self.assertRaisesRegex(RuntimeError, "before the persistent root thread"):
|
|
detach_session(session["session_id"])
|
|
finish_session(session["session_id"], exit_code=1, error="unused bootstrap")
|
|
|
|
self.assertEqual(launch_interactive(profile="codex-harness-team", cwd=box.workspace), 0)
|
|
active = next(
|
|
item for item in iter_sessions() if item["session_id"] != session["session_id"]
|
|
)
|
|
root_pid = active["root_pid"]
|
|
root_start_token = active["root_start_token"]
|
|
self.assertTrue(process_matches(root_pid, root_start_token))
|
|
with mock.patch.object(mmo_runtime, "terminate_root_host") as terminate:
|
|
detached = detach_session(active["session_id"])
|
|
terminate.assert_not_called()
|
|
self.assertEqual(detached["session"]["status"], "detached")
|
|
retained = load_session(active["session_id"])
|
|
self.assertEqual(retained["root_pid"], root_pid)
|
|
self.assertEqual(retained["root_start_token"], root_start_token)
|
|
self.assertTrue(process_matches(root_pid, root_start_token))
|
|
stopped = stop_session(active["session_id"], grace_seconds=0)
|
|
self.assertEqual(stopped["session"]["status"], "stopped")
|
|
self.assertFalse(process_matches(root_pid, root_start_token))
|
|
|
|
def test_root_app_server_host_is_owned_during_protocol_bootstrap(self) -> None:
|
|
with (
|
|
RuntimeSandbox() as box,
|
|
mock.patch.dict(
|
|
os.environ,
|
|
{"FAKE_CODEX_APP_SERVER_INITIALIZE_DELAY": "5"},
|
|
),
|
|
):
|
|
failures: list[BaseException] = []
|
|
|
|
def launch() -> None:
|
|
try:
|
|
run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Complete one bounded root task.",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(exc)
|
|
|
|
thread = threading.Thread(target=launch)
|
|
thread.start()
|
|
for _ in range(300):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and current.get("status") == "running"
|
|
and isinstance(current.get("root_pid"), int)
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("root app-server host was not published during bootstrap")
|
|
root_pid = current["root_pid"]
|
|
self.assertIsNone(current["root_thread_id"])
|
|
cancelled = cancel_session(current["session_id"])
|
|
self.assertEqual(cancelled["session"]["status"], "cancelled")
|
|
self.assertFalse(process_group_alive(root_pid))
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertTrue(failures)
|
|
|
|
def test_noninteractive_root_pending_input_detaches_for_same_thread_resume(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt="Ask before proceeding. FAKE_REQUEST_USER_INPUT",
|
|
wall_timeout_seconds=20,
|
|
)
|
|
self.assertEqual(result["status"], "detached", result)
|
|
self.assertEqual(
|
|
result["pending_requests"][0]["method"],
|
|
"item/tool/requestUserInput",
|
|
)
|
|
session_id = result["session"]["session_id"]
|
|
thread_id = result["session"]["root_thread_id"]
|
|
partial = box.state / "sessions" / session_id / "root-partial-result.md"
|
|
self.assertIn("requested operator input", partial.read_text(encoding="utf-8"))
|
|
current = load_session(session_id)
|
|
inspection = mmo_runtime.send_control_request(
|
|
Path(current["root_control_socket"]), {"action": "inspect"}
|
|
)["result"]
|
|
response = mmo_runtime.send_control_request(
|
|
Path(current["root_control_socket"]),
|
|
{
|
|
"action": "respond",
|
|
"expected_revision": inspection["control_revision"],
|
|
"request_id": result["pending_requests"][0]["id"],
|
|
"response": {"answers": {}},
|
|
},
|
|
)
|
|
self.assertTrue(response["result"]["responded"])
|
|
deadline = time.monotonic() + 10
|
|
while load_session(session_id)["status"] not in mmo_runtime.TERMINAL_SESSION_STATUSES:
|
|
self.assertLess(time.monotonic(), deadline)
|
|
time.sleep(0.05)
|
|
completed = load_session(session_id)
|
|
self.assertEqual(completed["root_thread_id"], thread_id)
|
|
self.assertEqual(completed["status"], "completed")
|
|
|
|
def test_graceful_stop_allows_app_server_worker_to_finalize(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="incident-hypothesis-triage", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="evidence_runner",
|
|
task_kind="evidence_collection",
|
|
task=(
|
|
"Return bounded evidence when the operator requests finalization. "
|
|
"FAKE_SLEEP_SECONDS=1"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
for _ in range(300):
|
|
if load_job(job["job_id"]).get("active_turn_id"):
|
|
break
|
|
time.sleep(0.02)
|
|
stopped = stop_session(session["session_id"], grace_seconds=5)
|
|
self.assertEqual(stopped["session"]["status"], "stopped")
|
|
self.assertEqual(load_job(job["job_id"])["status"], "completed")
|
|
|
|
def test_graceful_stop_steers_the_authoritative_active_root_turn(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
results: list[dict[str, Any]] = []
|
|
failures: list[BaseException] = []
|
|
|
|
def run() -> None:
|
|
try:
|
|
results.append(
|
|
run_root_exec(
|
|
profile="codex-harness-team",
|
|
cwd=box.workspace,
|
|
prompt=(
|
|
"Retain this root evidence until graceful finalization. "
|
|
"FAKE_SLEEP_SECONDS=2"
|
|
),
|
|
wall_timeout_seconds=20,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(exc)
|
|
|
|
thread = threading.Thread(target=run)
|
|
thread.start()
|
|
for _ in range(500):
|
|
sessions = iter_sessions()
|
|
current = sessions[0] if len(sessions) == 1 else None
|
|
if (
|
|
current is not None
|
|
and current.get("root_control_socket_ready")
|
|
and isinstance(current.get("active_root_turn_id"), str)
|
|
):
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
self.fail("root did not expose its authoritative active turn")
|
|
|
|
inspected = mmo_runtime.send_control_request(
|
|
Path(current["root_control_socket"]),
|
|
{"action": "inspect"},
|
|
)["result"]
|
|
self.assertEqual(inspected["active_turn_id"], current["active_root_turn_id"])
|
|
stopped = stop_session(current["session_id"], grace_seconds=8)
|
|
self.assertEqual(stopped["session"]["status"], "stopped")
|
|
thread.join(timeout=15)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertFalse(failures, failures)
|
|
self.assertEqual(results[0]["status"], "stopped")
|
|
|
|
directory = mmo_runtime.session_dir(current["session_id"])
|
|
event_rows = [
|
|
json.loads(line)
|
|
for line in (directory / "root-events.jsonl")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
]
|
|
sent_methods = [
|
|
row["message"].get("method")
|
|
for row in event_rows
|
|
if row.get("direction") == "sent" and isinstance(row.get("message"), dict)
|
|
]
|
|
self.assertIn("turn/steer", sent_methods)
|
|
self.assertEqual(sent_methods.count("turn/start"), 1)
|
|
audit_rows = [
|
|
json.loads(line)
|
|
for line in (directory / "audit.jsonl").read_text(encoding="utf-8").splitlines()
|
|
]
|
|
stop_event = next(
|
|
row for row in reversed(audit_rows) if row["event"] == "session_stopped"
|
|
)
|
|
self.assertTrue(stop_event["root_finalize_delivered"])
|
|
self.assertFalse(stop_event["root_forced"])
|
|
self.assertIn(
|
|
"FAKE_CODEX_OK",
|
|
(directory / "root-partial-result.md").read_text(encoding="utf-8"),
|
|
)
|
|
|
|
def test_result_accept_reject_and_patch_integration_are_explicit_and_audited(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
writable = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task=(
|
|
"Create one bounded generated module and return the required engineering "
|
|
"contract.\nFAKE_WRITE src/generated.py::value = 1"
|
|
),
|
|
mode="workspace-write",
|
|
write_scope_values=["src/generated.py"],
|
|
)
|
|
wait_for_jobs(
|
|
[writable["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
completed = load_job(writable["job_id"])
|
|
self.assertEqual(completed["status"], "completed", completed)
|
|
self.assertEqual(completed["result_state"], "unread")
|
|
self.assertFalse((box.workspace / "src" / "generated.py").exists())
|
|
self.assertEqual(completed["patch"]["changed_paths"], ["src/generated.py"])
|
|
self.assertEqual(
|
|
[item["relative_path"] for item in completed["artifacts"]],
|
|
["src/generated.py"],
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "result must be read"):
|
|
accept_result(
|
|
writable["job_id"],
|
|
"premature acceptance",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
read_result(writable["job_id"], session_id=session["session_id"])
|
|
accepted = accept_result(
|
|
writable["job_id"],
|
|
"focused patch and contract reviewed",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(accepted["result_state"], "accepted")
|
|
integrated = integrate_patch(
|
|
writable["job_id"],
|
|
"validated patch integrated into the canonical workspace",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(integrated["result_state"], "integrated")
|
|
self.assertEqual(
|
|
(box.workspace / "src" / "generated.py").read_text(encoding="utf-8"),
|
|
"value = 1\n",
|
|
)
|
|
|
|
advisory = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="flagship_escalation",
|
|
task_kind="review",
|
|
task="Review one bounded concern and return falsifiable evidence without writing.",
|
|
)
|
|
wait_for_jobs(
|
|
[advisory["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
read_result(advisory["job_id"], session_id=session["session_id"])
|
|
rejected = reject_result(
|
|
advisory["job_id"],
|
|
"evidence was not relevant to the accepted patch",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(rejected["result_state"], "rejected")
|
|
with self.assertRaisesRegex(RuntimeError, "result must be read"):
|
|
accept_result(
|
|
advisory["job_id"],
|
|
"attempted reversal",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
|
|
audit = [
|
|
json.loads(line)
|
|
for line in Path(session["audit_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
audited_events = {
|
|
row["event"]: row
|
|
for row in audit
|
|
if row["event"]
|
|
in {
|
|
"agent_result_accepted",
|
|
"agent_patch_integrated",
|
|
"agent_result_rejected",
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
set(audited_events),
|
|
{
|
|
"agent_result_accepted",
|
|
"agent_patch_integrated",
|
|
"agent_result_rejected",
|
|
},
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
row["run_id"] == session["current_run_id"]
|
|
for row in audited_events.values()
|
|
)
|
|
)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def _spawn_accepted_writer(self, session: Mapping[str, Any], path: str) -> dict[str, Any]:
|
|
writable = spawn_job(
|
|
session_id=str(session["session_id"]),
|
|
caller_agent=str(session["root_agent"]),
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task=(
|
|
"Create one bounded module and return the required engineering contract.\n"
|
|
f"FAKE_WRITE {path}::value = 1"
|
|
),
|
|
mode="workspace-write",
|
|
write_scope_values=[path],
|
|
)
|
|
wait_for_jobs(
|
|
[writable["job_id"]],
|
|
session_id=str(session["session_id"]),
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
read_result(writable["job_id"], session_id=str(session["session_id"]))
|
|
accept_result(
|
|
writable["job_id"],
|
|
"reviewed before the synthetic persistence failure",
|
|
session_id=str(session["session_id"]),
|
|
caller_agent=str(session["root_agent"]),
|
|
)
|
|
return writable
|
|
|
|
def test_patch_integration_rolls_back_when_lifecycle_persistence_fails(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
writable = self._spawn_accepted_writer(session, "src/rollback.py")
|
|
with (
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"publish_job_record",
|
|
side_effect=OSError("synthetic integration state failure"),
|
|
),
|
|
self.assertRaisesRegex(OSError, "synthetic integration state failure"),
|
|
):
|
|
integrate_patch(
|
|
writable["job_id"],
|
|
"exercise transactional integration rollback",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertFalse((box.workspace / "src" / "rollback.py").exists())
|
|
self.assertFalse((box.workspace / "src").exists())
|
|
self.assertEqual(load_job(writable["job_id"])["result_state"], "accepted")
|
|
self.assertFalse(load_session(session["session_id"]).get("tainted", False))
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_patch_integration_taints_session_when_rollback_fails(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
writable = self._spawn_accepted_writer(session, "src/rollback-failed.py")
|
|
job_directory = box.state / "jobs" / writable["job_id"]
|
|
real_publish_job_record = mmo_runtime.publish_job_record
|
|
|
|
def fail_integrated_state(directory: Path, data: Any) -> None:
|
|
if directory == job_directory and data.get("result_state") == "integrated":
|
|
raise OSError("synthetic integration state failure")
|
|
real_publish_job_record(directory, data)
|
|
|
|
with (
|
|
mock.patch.object(
|
|
mmo_runtime, "publish_job_record", side_effect=fail_integrated_state
|
|
),
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"reverse_applied_patch",
|
|
side_effect=RuntimeError("synthetic reverse failure"),
|
|
),
|
|
self.assertRaisesRegex(RuntimeError, "durably tainted.*manual"),
|
|
):
|
|
integrate_patch(
|
|
writable["job_id"],
|
|
"exercise failed rollback containment",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertEqual(
|
|
(box.workspace / "src" / "rollback-failed.py").read_text(encoding="utf-8"),
|
|
"value = 1\n",
|
|
)
|
|
self.assertEqual(load_job(writable["job_id"])["result_state"], "accepted")
|
|
persisted_session = load_session(session["session_id"])
|
|
self.assertTrue(persisted_session["tainted"])
|
|
self.assertEqual(
|
|
persisted_session["taint_reasons"][-1]["job_id"], writable["job_id"]
|
|
)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_patch_integration_reports_when_taint_persistence_also_fails(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
writable = self._spawn_accepted_writer(session, "src/taint-write-failed.py")
|
|
job_directory = box.state / "jobs" / writable["job_id"]
|
|
session_directory = box.state / "sessions" / session["session_id"]
|
|
real_publish_job_record = mmo_runtime.publish_job_record
|
|
real_publish_session_record = mmo_runtime.publish_session_record
|
|
|
|
def fail_job_state(directory: Path, data: Any) -> None:
|
|
if directory == job_directory and data.get("result_state") == "integrated":
|
|
raise OSError("synthetic integration state failure")
|
|
real_publish_job_record(directory, data)
|
|
|
|
def fail_session_state(
|
|
directory: Path,
|
|
data: Any,
|
|
*,
|
|
mirror_run: bool,
|
|
) -> None:
|
|
if directory == session_directory and data.get("tainted"):
|
|
raise OSError("synthetic taint persistence failure")
|
|
real_publish_session_record(directory, data, mirror_run=mirror_run)
|
|
|
|
with (
|
|
mock.patch.object(
|
|
mmo_runtime, "publish_job_record", side_effect=fail_job_state
|
|
),
|
|
mock.patch.object(
|
|
mmo_runtime, "publish_session_record", side_effect=fail_session_state
|
|
),
|
|
mock.patch.object(
|
|
mmo_runtime,
|
|
"reverse_applied_patch",
|
|
side_effect=RuntimeError("synthetic reverse failure"),
|
|
),
|
|
self.assertRaisesRegex(
|
|
RuntimeError, "taint marker also could not be persisted"
|
|
),
|
|
):
|
|
integrate_patch(
|
|
writable["job_id"],
|
|
"exercise failed durable containment",
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
)
|
|
self.assertTrue((box.workspace / "src" / "taint-write-failed.py").exists())
|
|
self.assertEqual(load_job(writable["job_id"])["result_state"], "accepted")
|
|
self.assertFalse(load_session(session["session_id"]).get("tainted", False))
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_invalid_advisory_contract_is_not_exposed_as_structured_result(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job_id = "20000101-000000-invalid-contract-0000000000"
|
|
directory = box.state / "jobs" / job_id
|
|
directory.mkdir(parents=True)
|
|
result_path = directory / "result.md"
|
|
result_path.write_text('{"plausible":"but invalid"}\n', encoding="utf-8")
|
|
metadata = {
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": job_id,
|
|
"session_id": session["session_id"],
|
|
"profile_id": session["profile_id"],
|
|
"snapshot_hash": session["snapshot_hash"],
|
|
"agent": "flagship_escalation",
|
|
"status": "completed_with_warnings",
|
|
"result_path": str(result_path),
|
|
"output_contract": {
|
|
"type": "object",
|
|
"required": ["required_field"],
|
|
},
|
|
"contract_enforcement": "warn",
|
|
"contract_valid": False,
|
|
"created_at": "2000-01-01T00:00:00+00:00",
|
|
"finished_at": "2000-01-01T00:00:01+00:00",
|
|
}
|
|
(directory / "metadata.json").write_text(
|
|
json.dumps(metadata) + "\n", encoding="utf-8"
|
|
)
|
|
result = read_result(job_id, session_id=session["session_id"])
|
|
self.assertEqual(result["content_format"], "text")
|
|
self.assertTrue(result["content"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_valid_json_null_result_remains_structured(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job_id = "20000101-000000-null-contract-000000000000"
|
|
directory = box.state / "jobs" / job_id
|
|
directory.mkdir(parents=True)
|
|
result_path = directory / "result.md"
|
|
structured_path = directory / "result.json"
|
|
result_path.write_text("null\n", encoding="utf-8")
|
|
structured_path.write_text("null\n", encoding="utf-8")
|
|
metadata = {
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": job_id,
|
|
"session_id": session["session_id"],
|
|
"run_id": session["current_run_id"],
|
|
"profile_id": session["profile_id"],
|
|
"snapshot_hash": session["snapshot_hash"],
|
|
"agent": "flagship_escalation",
|
|
"status": "completed",
|
|
"result_path": str(result_path),
|
|
"structured_result_path": str(structured_path),
|
|
"output_contract": {"type": "null"},
|
|
"contract_enforcement": "warn",
|
|
"contract_valid": True,
|
|
"result_kind": "final",
|
|
"result_state": "unread",
|
|
"created_at": "2000-01-01T00:00:00+00:00",
|
|
"finished_at": "2000-01-01T00:00:01+00:00",
|
|
}
|
|
(directory / "metadata.json").write_text(
|
|
json.dumps(metadata) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
result = read_result(job_id, session_id=session["session_id"])
|
|
|
|
self.assertEqual(result["content_format"], "json")
|
|
self.assertIsNone(result["content"])
|
|
self.assertEqual(result["total_chars"], len("null\n"))
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_result_pagination_is_lossless_and_mcp_hides_supervisor_paths(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job_id = "20000101-000000-paginated-result-000000000"
|
|
directory = box.state / "jobs" / job_id
|
|
directory.mkdir(parents=True)
|
|
result_path = directory / "result.md"
|
|
expected = "".join(f"{index:04d}:evidence\n" for index in range(650))
|
|
result_path.write_text(expected, encoding="utf-8")
|
|
metadata = {
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": job_id,
|
|
"session_id": session["session_id"],
|
|
"run_id": session["current_run_id"],
|
|
"profile_id": session["profile_id"],
|
|
"snapshot_hash": session["snapshot_hash"],
|
|
"agent": "flagship_escalation",
|
|
"status": "completed",
|
|
"result_path": str(result_path),
|
|
"events_path": str(directory / "events.jsonl"),
|
|
"stderr_path": str(directory / "stderr.log"),
|
|
"contract_valid": False,
|
|
"contract_enforcement": "warn",
|
|
"result_kind": "final",
|
|
"result_state": "unread",
|
|
"created_at": "2000-01-01T00:00:00+00:00",
|
|
"finished_at": "2000-01-01T00:00:01+00:00",
|
|
}
|
|
(directory / "metadata.json").write_text(
|
|
json.dumps(metadata) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
cursor = 0
|
|
pages: list[str] = []
|
|
observed_cursors: list[int] = []
|
|
while True:
|
|
page = read_result(
|
|
job_id,
|
|
session_id=session["session_id"],
|
|
cursor=cursor,
|
|
max_chars=700,
|
|
)
|
|
self.assertEqual(page["cursor"], cursor)
|
|
self.assertEqual(page["content_format"], "text")
|
|
self.assertNotIn("job", page)
|
|
self.assertNotIn("result_path", page)
|
|
pages.append(page["content"])
|
|
observed_cursors.append(cursor)
|
|
if page["next_cursor"] is None:
|
|
self.assertFalse(page["truncated"])
|
|
break
|
|
self.assertTrue(page["truncated"])
|
|
self.assertEqual(page["next_cursor"], cursor + len(page["content"]))
|
|
cursor = page["next_cursor"]
|
|
self.assertEqual("".join(pages), expected)
|
|
self.assertEqual(observed_cursors, sorted(set(observed_cursors)))
|
|
self.assertEqual(load_job(job_id)["result_state"], "read")
|
|
with self.assertRaisesRegex(ValueError, "exceeds the result length"):
|
|
read_result(
|
|
job_id,
|
|
session_id=session["session_id"],
|
|
cursor=len(expected) + 1,
|
|
)
|
|
|
|
waited = wait_for_jobs(
|
|
[job_id],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=0,
|
|
include_results=True,
|
|
)
|
|
preview = waited["results"][job_id]
|
|
self.assertTrue(preview["truncated"])
|
|
self.assertLessEqual(len(preview["preview"]), 4000)
|
|
self.assertIn("bounded previews", waited["result_reading_note"])
|
|
|
|
response = mcp_tool_call(
|
|
session,
|
|
session["root_agent"],
|
|
"agent_result",
|
|
{"job_id": job_id, "cursor": 0, "max_chars": 700},
|
|
)
|
|
self.assertEqual(response.returncode, 0, response.stderr)
|
|
row = next(
|
|
json.loads(line)
|
|
for line in response.stdout.splitlines()
|
|
if json.loads(line).get("id") == 2
|
|
)
|
|
encoded = json.dumps(row["result"], sort_keys=True)
|
|
for internal_name in (
|
|
"result_path",
|
|
"events_path",
|
|
"stderr_path",
|
|
"socket_path",
|
|
"full_result_path",
|
|
):
|
|
self.assertNotIn(internal_name, encoded)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_result_reader_rejects_state_paths_outside_the_canonical_job(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job_id = "20000101-000000-unsafe-result-path-0000000"
|
|
directory = box.state / "jobs" / job_id
|
|
directory.mkdir(parents=True)
|
|
result_path = directory / "result.md"
|
|
result_path.write_text("canonical result\n", encoding="utf-8")
|
|
external = box.root / "outside-result.md"
|
|
external.write_text("must not be disclosed\n", encoding="utf-8")
|
|
metadata = {
|
|
"schema_version": mmo_runtime.MMO_SCHEMA_VERSION,
|
|
"package_version": mmo_runtime.package_version(),
|
|
"job_id": job_id,
|
|
"session_id": session["session_id"],
|
|
"run_id": session["current_run_id"],
|
|
"profile_id": session["profile_id"],
|
|
"snapshot_hash": session["snapshot_hash"],
|
|
"agent": "flagship_escalation",
|
|
"status": "completed",
|
|
"result_path": str(external),
|
|
"created_at": "2000-01-01T00:00:00+00:00",
|
|
"finished_at": "2000-01-01T00:00:01+00:00",
|
|
}
|
|
metadata_path = directory / "metadata.json"
|
|
metadata_path.write_text(json.dumps(metadata) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "unsafe result path"):
|
|
read_result(job_id, session_id=session["session_id"])
|
|
|
|
metadata["result_path"] = str(result_path)
|
|
metadata["partial_result_path"] = str(external)
|
|
metadata_path.write_text(json.dumps(metadata) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "unsafe partial-result path"):
|
|
read_result(job_id, session_id=session["session_id"])
|
|
|
|
metadata.pop("partial_result_path")
|
|
metadata["contract_valid"] = True
|
|
metadata["structured_result_path"] = str(external)
|
|
metadata_path.write_text(json.dumps(metadata) + "\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "unsafe structured-result path"):
|
|
read_result(job_id, session_id=session["session_id"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_settings_codex_binary_is_pinned_for_root_and_workers(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
configured = Path(os.environ.pop("MMO_CODEX_BIN"))
|
|
settings_path = box.config / "settings.toml"
|
|
settings_path.write_text(
|
|
settings_path.read_text(encoding="utf-8").replace(
|
|
'codex_bin = "codex"', f'codex_bin = "{configured.as_posix()}"'
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
self.assertEqual(session["codex_binary"], str(configured.resolve()))
|
|
# A later settings change must not redirect descendants from the
|
|
# exact executable used to materialize this session.
|
|
settings_path.write_text(
|
|
settings_path.read_text(encoding="utf-8").replace(
|
|
configured.as_posix(), (box.root / "missing-codex").as_posix()
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="literal_scout",
|
|
literal_task={
|
|
"operation": "summarize_supplied",
|
|
"text": "README.md literal evidence fixture",
|
|
},
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
self.assertEqual(load_job(job["job_id"])["status"], "completed")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_builtin_auth_uses_operator_link_mode_and_excludes_mcp_oauth(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
oauth = box.base_codex_home / ".credentials.json"
|
|
oauth.write_text('{"unrelated":"mcp-oauth"}\n', encoding="utf-8")
|
|
settings_path = box.config / "settings.toml"
|
|
settings_path.write_text(
|
|
settings_path.read_text(encoding="utf-8").replace(
|
|
'auth_link_mode = "shared"', 'auth_link_mode = "copy"'
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
session = create_session(profile="codex-harness-team", cwd=box.workspace)
|
|
try:
|
|
home = Path(session["homes"][session["root_agent"]]["home"])
|
|
auth = home / "auth.json"
|
|
self.assertTrue(auth.is_file())
|
|
self.assertFalse(auth.is_symlink())
|
|
self.assertEqual(auth.stat().st_mode & 0o777, 0o600)
|
|
self.assertFalse((home / ".credentials.json").exists())
|
|
config = tomllib.loads((home / "config.toml").read_text(encoding="utf-8"))
|
|
self.assertEqual(config["cli_auth_credentials_store"], "file")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_low_trust_and_write_scope_conflict_enforcement(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
destination = clone_profile("access-efficient-escalation-lab", "scope-conflict-test")
|
|
profile = read_toml(destination / "profile.toml")
|
|
profile["agents"]["routine_engineer"]["max_active"] = 2
|
|
profile["coordination"]["max_active_writers"] = 2
|
|
(destination / "profile.toml").write_text(toml_dumps(profile), encoding="utf-8")
|
|
session = create_access_lab_session(
|
|
profile="scope-conflict-test",
|
|
cwd=box.workspace,
|
|
)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
with self.assertRaises(PermissionError):
|
|
spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="literal_scout",
|
|
literal_task={
|
|
"operation": "locate",
|
|
"needle": "README.md",
|
|
"paths": ["."],
|
|
},
|
|
mode="workspace-write",
|
|
write_scope_values=["README.md"],
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "low-trust agents accept literal_task"):
|
|
spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="literal_scout",
|
|
task_kind="architecture",
|
|
task="Perform a complete architecture analysis of this repository.",
|
|
mode="read-only",
|
|
)
|
|
first = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task="Make a bounded isolated change. FAKE_SLEEP=3",
|
|
mode="workspace-write",
|
|
write_scope_values=["src"],
|
|
)
|
|
with self.assertRaises(RuntimeError):
|
|
spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task="Make a second bounded isolated change while the first job is active.",
|
|
mode="workspace-write",
|
|
write_scope_values=["src/module.py"],
|
|
)
|
|
cancel_job(
|
|
first["job_id"], session_id=session["session_id"], reason="duplicate test work"
|
|
)
|
|
wait_for_jobs(
|
|
[first["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
self.assertEqual(load_job(first["job_id"])["cancel_reason"], "duplicate test work")
|
|
self.assertEqual(load_job(first["job_id"])["status"], "cancelled")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_terminal_jobs_release_capacity_for_sequential_role_reuse(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
completed_ids: list[str] = []
|
|
for index in range(5):
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="test",
|
|
task=f"Inspect bounded concern {index} and report factual evidence.",
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
self.assertIn(
|
|
load_job(job["job_id"])["status"],
|
|
{"completed", "completed_with_warnings", "failed"},
|
|
)
|
|
completed_ids.append(job["job_id"])
|
|
self.assertEqual(len(set(completed_ids)), 5)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_optional_write_scope_becomes_whole_delegated_workspace(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
destination = clone_profile("access-efficient-escalation-lab", "implicit-write-scope")
|
|
profile = read_toml(destination / "profile.toml")
|
|
profile["agents"]["routine_engineer"]["write_scope_required"] = False
|
|
(destination / "profile.toml").write_text(toml_dumps(profile), encoding="utf-8")
|
|
session = create_session(profile="implicit-write-scope", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task="Make a bounded test change with the profile's implicit scope.",
|
|
mode="workspace-write",
|
|
)
|
|
wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["write_scope"], ["."])
|
|
self.assertEqual(final["write_scope_resolved"], [str(box.workspace)])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_mcp_cancel_forwards_operator_reason(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
malformed = mcp_tool_call(
|
|
session,
|
|
session["root_agent"],
|
|
"agent_status",
|
|
{"unexpected": True},
|
|
)
|
|
malformed_row = json.loads(malformed.stdout.splitlines()[-1])
|
|
self.assertEqual(malformed_row["error"]["code"], -32602)
|
|
self.assertEqual(malformed_row["error"]["message"], "invalid params")
|
|
self.assertIn("invalid tool arguments", malformed_row["error"]["data"])
|
|
unknown = mcp_tool_call(
|
|
session,
|
|
session["root_agent"],
|
|
"not_an_advertised_tool",
|
|
{},
|
|
)
|
|
unknown_row = json.loads(unknown.stdout.splitlines()[-1])
|
|
self.assertEqual(unknown_row["error"]["code"], -32602)
|
|
self.assertIn("unknown tool", unknown_row["error"]["data"])
|
|
execution_failure = mcp_tool_call(
|
|
session,
|
|
session["root_agent"],
|
|
"agent_result",
|
|
{"job_id": "20000101-000000-missing-result-0000000000"},
|
|
)
|
|
failure_row = json.loads(execution_failure.stdout.splitlines()[-1])
|
|
self.assertTrue(failure_row["result"]["isError"])
|
|
self.assertIn("FileNotFoundError", failure_row["result"]["content"][0]["text"])
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="flagship_escalation",
|
|
task_kind="analysis",
|
|
task="Analyze slowly so cancellation can be tested. FAKE_SLEEP=3",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "at most 500"):
|
|
cancel_job(
|
|
job["job_id"],
|
|
session_id=session["session_id"],
|
|
reason="x" * 501,
|
|
)
|
|
self.assertIn(load_job(job["job_id"])["status"], {"queued", "running"})
|
|
response = mcp_tool_call(
|
|
session,
|
|
session["root_agent"],
|
|
"agent_cancel",
|
|
{"job_id": job["job_id"], "reason": "superseded through MCP"},
|
|
)
|
|
self.assertEqual(response.returncode, 0, response.stderr)
|
|
self.assertNotIn('"error"', response.stdout)
|
|
wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=10,
|
|
include_results=False,
|
|
)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "cancelled")
|
|
self.assertEqual(final["cancel_reason"], "superseded through MCP")
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_smoke_runner_exercises_native_and_mcp_backends(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
(box.workspace / "README.md").write_text("fixture\n", encoding="utf-8")
|
|
result = smoke_profile(
|
|
"adaptive-engineering",
|
|
cwd=str(box.workspace),
|
|
bindings={},
|
|
root_only=False,
|
|
workers_only=True,
|
|
)
|
|
self.assertTrue(result["passed"], result)
|
|
by_agent = {item["agent"]: item for item in result["results"]}
|
|
self.assertEqual(by_agent["repo_scout"]["backend"], "native")
|
|
self.assertTrue(by_agent["repo_scout"]["marker_present"])
|
|
self.assertEqual(by_agent["implementation_specialist"]["backend"], "mcp")
|
|
|
|
def test_root_harness_prompt_terminates_only_goal_mode(self) -> None:
|
|
turn_prompt = _root_harness_prompt("Do the bounded check.", "turn")
|
|
self.assertEqual(turn_prompt, "Do the bounded check.")
|
|
goal_prompt = _root_harness_prompt("Do the bounded check.", "goal")
|
|
self.assertIn("call `update_goal`", goal_prompt)
|
|
self.assertIn('`status="complete"`', goal_prompt)
|
|
self.assertIn("final message alone does not terminate an active goal", goal_prompt)
|
|
|
|
def test_root_waits_for_terminal_turn_after_goal_completion(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
result = run_root_exec(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
prompt="Complete the bounded probe. FAKE_GOAL_COMPLETE_BEFORE_TURN",
|
|
wall_timeout_seconds=20,
|
|
sandbox_mode="read-only",
|
|
)
|
|
self.assertEqual(result["status"], "completed", result)
|
|
self.assertEqual(result["exit_code"], 0, result)
|
|
self.assertIn("FAKE_CODEX_OK", result["result"])
|
|
history = read_json(
|
|
mmo_runtime.session_dir(result["session"]["session_id"])
|
|
/ "root-terminal-history.json"
|
|
)
|
|
self.assertEqual(history["turns"][-1]["status"], "completed")
|
|
|
|
def test_long_goal_objectives_preserve_complete_root_and_worker_turns(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
root_tail = "ROOT_FULL_PROMPT_TAIL"
|
|
root_prompt = (
|
|
"Complete this bounded root probe. FAKE_GOAL_COMPLETE_BEFORE_TURN\n"
|
|
+ ("root-context-Ω\n" * 340)
|
|
+ root_tail
|
|
)
|
|
self.assertGreater(len(root_prompt), 4000)
|
|
result = run_root_exec(
|
|
profile="adaptive-engineering",
|
|
cwd=box.workspace,
|
|
prompt=root_prompt,
|
|
wall_timeout_seconds=20,
|
|
sandbox_mode="read-only",
|
|
)
|
|
self.assertEqual(result["status"], "completed", result)
|
|
root_session = load_session(result["session"]["session_id"])
|
|
self.assertEqual(len(root_session["root_goal_objective"]), 4000)
|
|
self.assertIn("sha256=", root_session["root_goal_objective"])
|
|
root_events = Path(result["events_path"]).read_text(encoding="utf-8")
|
|
self.assertIn(root_tail, root_events)
|
|
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
worker_tail = "WORKER_FULL_TASK_TAIL"
|
|
task = (
|
|
"Analyze this bounded invariant without modifying files. "
|
|
"FAKE_GOAL_COMPLETE_BEFORE_TURN\n" + ("worker-context-λ\n" * 330) + worker_tail
|
|
)
|
|
self.assertGreater(len(task), 4000)
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task=task,
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertEqual(len(final["goal_objective"]), 4000)
|
|
self.assertIn("sha256=", final["goal_objective"])
|
|
self.assertEqual(final["task"], task)
|
|
self.assertIn(
|
|
worker_tail,
|
|
(mmo_runtime.job_dir(job["job_id"]) / "prompt.txt").read_text(encoding="utf-8"),
|
|
)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_goal_worker_waits_for_terminal_turn_before_strict_finalization(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task=(
|
|
"Inspect one bounded invariant without writing. "
|
|
"FAKE_GOAL_COMPLETE_BEFORE_TURN"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertTrue(final["contract_valid"], final)
|
|
history = read_json(mmo_runtime.job_dir(job["job_id"]) / "terminal-history.json")
|
|
self.assertEqual(history["turns"][-1]["status"], "completed")
|
|
events = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 2, starts)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_goal_finalization_recovers_after_persisted_turn_loses_transport(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
session = create_session(profile="adaptive-engineering", cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="implementation_specialist",
|
|
task_kind="analysis",
|
|
task=(
|
|
"Inspect one bounded invariant without writing. "
|
|
"FAKE_GOAL_COMPLETE_BEFORE_TURN "
|
|
"FAKE_APP_SERVER_CRASH_DURING_FINALIZATION_ONCE"
|
|
),
|
|
mode="read-only",
|
|
)
|
|
waited = wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
self.assertFalse(waited["unfinished"], waited)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "completed", final)
|
|
self.assertTrue(final["contract_valid"], final)
|
|
self.assertGreaterEqual(final["recovery_attempts"], 1)
|
|
events = [
|
|
json.loads(line)
|
|
for line in Path(final["events_path"]).read_text(encoding="utf-8").splitlines()
|
|
]
|
|
starts = [
|
|
row
|
|
for row in events
|
|
if row.get("direction") == "sent"
|
|
and row.get("message", {}).get("method") == "turn/start"
|
|
]
|
|
self.assertEqual(len(starts), 2, starts)
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
def test_smoke_stops_a_root_detached_by_its_external_harness(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
root_result = {
|
|
"session": {"session_id": "smoke-detached-session"},
|
|
"status": "detached",
|
|
"root_status": "harness_wall_detached",
|
|
"exit_code": 124,
|
|
"elapsed_seconds": 1.0,
|
|
"events_path": "",
|
|
"result": "partial evidence",
|
|
}
|
|
cleanup = {"session": {"status": "stopped"}}
|
|
with (
|
|
mock.patch("mmo_diagnostics.run_root_exec", return_value=root_result),
|
|
mock.patch("mmo_diagnostics.stop_session", return_value=cleanup) as stop,
|
|
):
|
|
result = smoke_profile(
|
|
"codex-harness-team",
|
|
cwd=str(box.workspace),
|
|
bindings={},
|
|
root_only=True,
|
|
workers_only=False,
|
|
)
|
|
self.assertFalse(result["passed"])
|
|
stop.assert_called_once_with("smoke-detached-session", grace_seconds=0)
|
|
self.assertEqual(
|
|
result["results"][0]["cleanup"],
|
|
{"mode": "graceful_stop", "result": cleanup},
|
|
)
|
|
|
|
def test_out_of_scope_git_audit_including_preexisting_dirty_file(self) -> None:
|
|
with RuntimeSandbox() as box:
|
|
box.init_git()
|
|
dirty = box.workspace / "README.md"
|
|
dirty.write_text("preexisting dirty\n", encoding="utf-8")
|
|
session = create_access_lab_session(cwd=box.workspace)
|
|
mark_session_running(session["session_id"], os.getpid())
|
|
try:
|
|
job = spawn_job(
|
|
session_id=session["session_id"],
|
|
caller_agent=session["root_agent"],
|
|
caller_job_id=None,
|
|
agent_id="routine_engineer",
|
|
task_kind="implement",
|
|
task="Make the requested scoped change.\nFAKE_WRITE README.md::changed outside authorized scope",
|
|
mode="workspace-write",
|
|
write_scope_values=["src/allowed.py"],
|
|
)
|
|
wait_for_jobs(
|
|
[job["job_id"]],
|
|
session_id=session["session_id"],
|
|
timeout_seconds=20,
|
|
include_results=False,
|
|
)
|
|
final = load_job(job["job_id"])
|
|
self.assertEqual(final["status"], "failed")
|
|
self.assertIn("out-of-scope mutation", final["error"])
|
|
persisted = load_session(session["session_id"])
|
|
self.assertTrue(persisted["tainted"])
|
|
finally:
|
|
finish_session(session["session_id"], exit_code=0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|