test_interactive_chat_native_integration_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Actual shared-chat adapters with isolated native seams, never server startup."""
import ast
import asyncio
from contextlib import asynccontextmanager
import hashlib
import json
from pathlib import Path
import re
import tempfile
import time
from types import MethodType, SimpleNamespace
import unittest
from unittest.mock import AsyncMock, Mock, patch
from fastapi import HTTPException
from fastapi.responses import Response
from interactive_chat_controls import ChatControlError, InteractiveChatControls
from interactive_chat_native import shared_events, shared_native_value, shared_session
from test_interactive_chat_controls import native_models
from test_interactive_chat_integration_isolated import load_glue
from test_shared_chat_video_native_integration_isolated import install_media_glue, register_synthetic_video
def load_native_glue():
namespace, tree = load_glue()
install_media_glue(namespace, tree)
names = {"interactive_chat_native_page", "interactive_chat_native_snapshot",
"steer_interactive_chat_prompt", "run_interactive_chat_job", "control_interactive_chat",
"get_cross_chat_handoff", "public_cross_chat_envelope", "is_async_route_message",
"async_route_conversation_fields", "async_message_target_fields"}
nodes = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names]
assert {node.name for node in nodes} == names
for node in nodes:
node.decorator_list = []
job_store = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "JobStore")
manual = next(node for node in job_store.body if isinstance(node, ast.AsyncFunctionDef) and node.name == "request_manual_run")
namespace.update(native_models())
namespace.update(asyncio=asyncio, time=time, json=json, hashlib=hashlib, Response=Response,
shared_events=shared_events, shared_native_value=shared_native_value, shared_session=shared_session,
ChatControlError=ChatControlError, InteractiveChatControls=InteractiveChatControls)
exec(compile(ast.fix_missing_locations(ast.Module(body=[*nodes, manual], type_ignores=[])), "<isolated-native-chat-adapters>", "exec"), namespace)
return namespace
class InteractiveChatNativeGlueTests(unittest.IsolatedAsyncioTestCase):
@classmethod
def setUpClass(cls):
cls.native = load_native_glue()
def setUp(self):
self.share = "interactive_" + "a" * 32
self.request = "request_0000000001"
self.lock = asyncio.Lock()
self.row = {"id": "event-one", "session_id": "chat-one", "seq": 1,
"type": "reasoning_summary", "phase": "commentary", "text": "Progress",
"ts": "2026-09-12T00:00:00.123Z", "run_id": "native-run"}
self.native.update({
"interactive_chat_session_available": lambda sid: sid == "chat-one",
"STORE": SimpleNamespace(sessions={"chat-one": {"id": "chat-one", "title": "Shared", "backend": "codex", "cwd": "/private/owner"}}),
"BUSY_SESSIONS": {"chat-one"}, "ACTIVE": {},
"read_semantic_timeline_page": Mock(return_value={"events": [self.row, {"id": "file-one", "session_id": "chat-one", "type": "file_uploaded", "seq": 2}], "semantic_total": 2, "semantic_omitted_before": 1, "next_semantic_before": 1}),
"public_session": lambda value: dict(value), "DEFAULT_BACKEND": "codex",
"BACKEND_CODEX": "codex", "BACKEND_CLAUDE": "claude",
"CODEX_TRANSPORT": "app-server", "CODEX_TRANSPORT_EXEC": "exec",
"CLAUDE_TRANSPORT": "sdk", "CLAUDE_TRANSPORT_PRINT": "print",
"CODEX_INTERACTIVE_CLIENT_CAPABILITY": "codex-test", "CLAUDE_SDK_INTERACTIVE_CLIENT_CAPABILITY": "claude-test",
"CODEX_GOALS_ENABLED": True, "CODEX_DEFAULT_APPROVAL_POLICY": "on-request",
"AGENT_TOKEN": "synthetic-native-admin",
"PROVIDER_JOBS_ACCESS_MODES": ("full", "read_only", "blocked"),
"PROVIDER_JOBS_ACCESS_DEFAULT": "full",
"PROVIDER_CROSS_CHAT_ROUTE_PAIR_ID_RE": re.compile(r"^pair_[0-9a-f]{32}$"),
"sanitized_provider_route_label": lambda value: str(value or ""),
"CODEX_DEFAULT_SANDBOX_MODE": "workspace-write", "CODEX_DEFAULT_PERMISSION_PROFILE": None,
"CODEX_DEFAULT_APPROVALS_REVIEWER": "user", "CLAUDE_PERMISSION_MODE_OPTIONS": ("default",),
"CLAUDE_STOP_FENCE_SESSIONS": set(), "effective_claude_permission_mode": lambda _: "default",
"session_provider_id": lambda _: "synthetic-native-thread", "claude_provider_id_for_session": lambda _: None,
"CODEX_PENDING_INTERACTIONS_LOCK": asyncio.Lock(), "CLAUDE_PENDING_INTERACTIONS_LOCK": asyncio.Lock(),
"CODEX_PENDING_INTERACTIONS": {"own": {"id": "approval-own", "session_id": "chat-one", "params": {"headers": {"token": "private"}}}, "foreign": {"id": "approval-foreign", "session_id": "chat-two"}},
"CLAUDE_PENDING_INTERACTIONS": {}, "public_codex_interaction": lambda row: dict(row), "public_claude_interaction": lambda row: dict(row),
"get_codex_goal": AsyncMock(return_value={"enabled": True, "goal": None}),
"queued_turns_snapshot": AsyncMock(return_value=[{"queued_id": "queued-own", "prompt": "Waiting", "file_ids": ["private-file"]}]),
"strip_agentsdock_generated_user_text": lambda text, **kwargs: text,
"list_session_jobs": AsyncMock(return_value={"jobs": []}),
"session_lifecycle_lock": lambda sid: self.lock, "ensure_session_not_deleting": Mock(),
"INTERACTIVE_CHAT_LIVE": SimpleNamespace(notify=Mock()),
"INTERACTIVE_CHAT_CATALOG": None, "INTERACTIVE_CHAT_CATALOG_LOCK": asyncio.Lock(),
"runtime_catalog": AsyncMock(return_value={"backends": {"codex": {"models": [], "cwd": "/private"}}}),
})
for name in ("post_turn", "post_run_queued_turn_now", "stop_turn_endpoint", "patch_queued_turn", "delete_queued_turn",
"post_move_queued_turn", "update_session", "put_codex_goal", "delete_codex_goal", "create_session_job",
"update_session_job", "delete_session_job", "post_codex_interaction_response", "post_claude_interaction_response"):
self.native[name] = AsyncMock(return_value={"ok": True})
async def test_indexed_page_preserves_identity_and_rejects_foreign_rows(self):
result = await self.native["interactive_chat_native_page"]("chat-one", semantic_before=20, limit=25)
self.assertEqual(result["events"], [self.row])
self.assertTrue(result["semantic_paging"])
self.native["read_semantic_timeline_page"].assert_called_once_with("chat-one", semantic_before=20, limit=25)
self.native["read_semantic_timeline_page"].return_value = {"events": [{**self.row, "session_id": "chat-two"}]}
with self.assertRaises(ValueError): await self.native["interactive_chat_native_page"]("chat-one")
async def test_native_snapshot_is_one_chat_without_runtime_probe_or_private_file_fields(self):
result = await self.native["interactive_chat_native_snapshot"]("chat-one")
self.assertEqual(result["events"], [self.row])
self.assertNotIn("cwd", result["session"])
self.assertEqual(result["queue"][0]["queued_id"], "queued-own")
self.assertEqual(result["queue"][0]["file_ids"], [])
self.assertEqual([row["id"] for row in result["codex_runtime"]["pending_interactions"]], ["approval-own"])
self.assertNotIn("headers", result["codex_runtime"]["pending_interactions"][0]["params"])
self.assertFalse(result["health"]["capabilities"]["workspace_files"]["available"])
self.native["runtime_catalog"].assert_not_awaited()
self.native["list_session_jobs"].assert_awaited_once_with("chat-one")
async def test_native_snapshot_preserves_jobs_policy_and_actual_capability(self):
for policy in ("full", "read_only", "blocked"):
with self.subTest(policy=policy):
self.native["STORE"].sessions["chat-one"]["provider_jobs_access"] = policy
result = await self.native["interactive_chat_native_snapshot"]("chat-one")
self.assertEqual(result["session"]["provider_jobs_access"], policy)
self.assertEqual(result["health"]["capabilities"]["provider_jobs_access_control_v1"], {
"available": True, "version": 1, "modes": ["full", "read_only", "blocked"], "default": "full",
})
self.native.update(AGENT_TOKEN="", PROVIDER_JOBS_ACCESS_DEFAULT="blocked")
result = await self.native["interactive_chat_native_snapshot"]("chat-one")
capability = result["health"]["capabilities"]["provider_jobs_access_control_v1"]
self.assertFalse(capability["available"])
self.assertEqual(capability["default"], "blocked")
self.native["runtime_catalog"].assert_not_awaited()
async def test_native_page_issues_real_media_and_discards_event_supplied_handles(self):
with tempfile.TemporaryDirectory(prefix="native-page-video-") as directory:
root = Path(directory)
own = register_synthetic_video(root)
other = register_synthetic_video(root, "e", "chat-two")
self.native["FILES_ROOT"] = root
fake = {"id": "video_ZmFrZQ." + "a" * 64, "filename": "fake.mp4", "content_type": "video/mp4", "size": 1}
self.native["read_semantic_timeline_page"].return_value = {"events": [
{"id": "sent", "session_id": "chat-one", "type": "turn_started", "prompt": "Sent",
"file_ids": [own], "shared_videos": [fake]},
{"id": "artifact", "session_id": "chat-one", "type": "artifact_created",
"artifact": {"id": own, "path": "/private/owner"}},
{"id": "foreign", "session_id": "chat-one", "type": "turn_started", "prompt": "Other",
"file_ids": [other]},
{"id": "raw", "session_id": "chat-one", "type": "assistant_text", "text": "Visible",
"shared_videos": [fake]},
{"id": "upload", "session_id": "chat-one", "type": "file_uploaded", "file": {"id": own}},
]}
result = await self.native["interactive_chat_native_page"]("chat-one")
rows = {row["id"]: row for row in result["events"]}
self.assertEqual(set(rows), {"sent", "artifact", "foreign", "raw"})
self.assertEqual(rows["sent"]["shared_videos"], rows["artifact"]["shared_videos"])
self.assertNotEqual(rows["sent"]["shared_videos"][0]["id"], fake["id"])
self.assertNotIn("shared_videos", rows["foreign"])
self.assertNotIn("shared_videos", rows["raw"])
self.assertNotIn("file_ids", rows["sent"])
self.assertNotIn("artifact", rows["artifact"])
self.assertNotIn("/private/owner", json.dumps(result))
async def test_handoff_read_uses_native_body_revision_and_participant_visibility(self):
original = "Original synthetic message.\n" * 180
edited = "Recipient-edited synthetic message.\n" * 160
for source, target in (("chat-one", "chat-two"), ("chat-two", "chat-one")):
with self.subTest(source=source, target=target):
record = {"id": "handoff-synthetic", "kind": "instruction", "source_session_id": source,
"target_session_id": target, "body": original, "target_body": edited, "message_revision": 2,
"authorization_kind": "configured_route", "authorization_route_id": "route-not-for-guest",
"authorization_pair_id": "pair_" + "a" * 32, "delivery_mode": "mailbox", "status": "stored"}
ledger = SimpleNamespace(get=AsyncMock(return_value=record),
mailbox_envelopes=AsyncMock(return_value=[{**record, "read_at": "2026-09-12T12:00:00Z"}]))
self.native["CROSS_CHAT"] = ledger
result = await self.native["control_interactive_chat"]("chat-one", "handoffs.get", {"id": record["id"]})
handoff = result["handoff"]
self.assertEqual(handoff["id"], record["id"])
self.assertEqual(handoff["message_id"], record["id"])
self.assertEqual(handoff["conversation_id"], record["authorization_pair_id"])
self.assertEqual(handoff["body"], original)
self.assertEqual(handoff["body_sha256"], hashlib.sha256(original.encode()).hexdigest())
self.assertEqual(handoff["inbox_state"], "read")
for key in ("authorization_kind", "authorization_route_id"):
self.assertNotIn(key, handoff)
if target == "chat-one":
self.assertEqual(handoff["target_body"], edited)
self.assertEqual(handoff["message_revision"], 2)
self.assertTrue(handoff["message_edited_by_user"])
else:
for key in ("target_body", "message_revision", "message_edited_by_user"):
self.assertNotIn(key, handoff)
self.assertEqual(ledger.get.await_count, 2)
ledger.mailbox_envelopes.assert_awaited_once_with(message_id=record["id"])
self.native["post_turn"].assert_not_awaited()
self.native["INTERACTIVE_CHAT_LIVE"].notify.assert_not_called()
async def test_handoff_read_denies_missing_foreign_or_changed_identity(self):
owned = {"id": "handoff-synthetic", "source_session_id": "chat-one", "target_session_id": "chat-two"}
for record in (None, {**owned, "source_session_id": "chat-three"}, {**owned, "id": "wrong-envelope"}):
with self.subTest(record=record):
self.native["CROSS_CHAT"] = SimpleNamespace(get=AsyncMock(return_value=record))
projection = AsyncMock()
with patch.dict(self.native, get_cross_chat_handoff=projection):
with self.assertRaises(ChatControlError) as denied:
await self.native["control_interactive_chat"]("chat-one", "handoffs.get", {"id": owned["id"]})
self.assertEqual(denied.exception.code, "forbidden")
projection.assert_not_awaited()
self.native["CROSS_CHAT"] = SimpleNamespace(get=AsyncMock(return_value=owned))
projection = AsyncMock(return_value={"handoff": {**owned, "target_session_id": "chat-three", "body": "not returned"}})
with patch.dict(self.native, get_cross_chat_handoff=projection):
with self.assertRaises(ChatControlError):
await self.native["control_interactive_chat"]("chat-one", "handoffs.get", {"id": owned["id"]})
for payload in ({}, {"id": owned["id"], "session_id": "chat-two"}, {"id": [owned["id"]]}):
self.native["CROSS_CHAT"].get.reset_mock()
with self.assertRaises(ChatControlError):
await self.native["control_interactive_chat"]("chat-one", "handoffs.get", payload)
self.native["CROSS_CHAT"].get.assert_not_awaited()
async def test_control_uses_exact_native_model_and_rejects_paths_before_mutation(self):
self.native["patch_queued_turn"].return_value = {"ok": True, "file_ids": ["private"]}
result = await self.native["control_interactive_chat"]("chat-one", "queue.edit", {"id": "queued-own", "prompt": "Edited", "expected_message_revision": 3})
self.assertEqual(result, {"accepted": True, "result": {"ok": True}})
sid, qid, request = self.native["patch_queued_turn"].await_args.args
self.assertEqual((sid, qid), ("chat-one", "queued-own"))
self.assertEqual(request.model_dump(exclude_unset=True), {"prompt": "Edited", "expected_message_revision": 3})
with self.assertRaises(ChatControlError) as denied:
await self.native["control_interactive_chat"]("chat-one", "settings.update", {"cwd": "/other"})
self.assertEqual(denied.exception.code, "invalid_request")
self.native["update_session"].assert_not_awaited()
with self.assertRaises(HTTPException): await self.native["control_interactive_chat"]("chat-two", "turn.stop", {})
self.native["stop_turn_endpoint"].assert_not_awaited()
async def test_steer_preserves_trusted_author_and_promotes_only_receipted_queue_id(self):
self.native["post_turn"].return_value = {"queued": True, "queued_id": "queued-exact"}
await self.native["steer_interactive_chat_prompt"]("chat-one", "A human clarification", share_id=self.share, request_id=self.request)
sid, request = self.native["post_turn"].await_args.args
self.assertEqual(sid, "chat-one")
self.assertEqual(request.shared_chat_metadata, {"shared_chat_id": self.share, "shared_chat_request_id": self.request, "author_label": "Collaborator"})
self.assertEqual(request.chat_references, [])
sid, qid, request = self.native["post_run_queued_turn_now"].await_args.args
self.assertEqual((sid, qid), ("chat-one", "queued-exact"))
self.assertTrue(request.accept_deferred_queue_response)
self.native["stop_turn_endpoint"].assert_not_awaited()
for receipt in ({"queued": True}, {}, {"queued": "false"}):
self.native["post_turn"].return_value = receipt
self.native["post_run_queued_turn_now"].reset_mock()
with self.subTest(receipt=receipt), self.assertRaises(HTTPException):
await self.native["steer_interactive_chat_prompt"]("chat-one", "Text", share_id=self.share, request_id=self.request)
self.native["post_run_queued_turn_now"].assert_not_awaited()
async def test_job_run_releases_lifecycle_lock_before_normal_turn_admission(self):
async def dispatch(_jid):
async with self.lock: # Actual _start_job_run calls start_turn, which takes this lock.
return {"queued": False, "run_id": "synthetic-run"}
jobs = SimpleNamespace(jobs={"job-own": {"id": "job-own", "session_id": "chat-one"}}, _lock=asyncio.Lock(),
save=AsyncMock(), _dispatch_pending_manual_run=dispatch, pause_for_session=AsyncMock())
jobs.request_manual_run = MethodType(self.native["request_manual_run"], jobs)
self.native.update(JOBS=jobs, now_iso=lambda: "2026-09-12T00:00:00Z", new_job_revision=lambda: "revision",
event_job=lambda row: dict(row), append_event=AsyncMock())
result = await asyncio.wait_for(self.native["run_interactive_chat_job"]("chat-one", "job-own"), 1)
self.assertEqual(result["run_id"], "synthetic-run")
jobs.save.assert_awaited_once()
with self.assertRaises(HTTPException):
await jobs.request_manual_run("job-own", expected_session_id="chat-two")
self.assertEqual(jobs.save.await_count, 1)
@asynccontextmanager
async def changed_owner():
jobs.jobs["job-own"] = {"id": "job-own", "session_id": "chat-two"}
yield
jobs._lock = changed_owner()
with self.assertRaises(HTTPException):
await jobs.request_manual_run("job-own", expected_session_id="chat-one")
self.assertEqual(jobs.save.await_count, 1)
async def test_explicit_catalog_read_is_cached_and_browser_cannot_select_other_chat(self):
for _ in range(2):
result = await self.native["control_interactive_chat"]("chat-one", "runtime.catalog", {})
self.assertEqual(result, {"backends": {"codex": {"models": []}}})
self.native["runtime_catalog"].assert_awaited_once_with()
with self.assertRaises(ChatControlError):
await self.native["control_interactive_chat"]("chat-one", "timeline.older", {"session_id": "chat-two"})
self.native["read_semantic_timeline_page"].assert_not_called()
if __name__ == "__main__":
unittest.main()