diff --git a/docs/company-metadata-ownership.md b/docs/company-metadata-ownership.md index 5f94c34..7d059f6 100644 --- a/docs/company-metadata-ownership.md +++ b/docs/company-metadata-ownership.md @@ -11,7 +11,7 @@ The executable owner matrix lives in `opc/layer2_organization/metadata_ownership | Owner | Examples | Rule | |---|---|---| -| `work_item` | `work_kind`, `current_turn_mode`, dependency/waiting ids, handoff/context preview, prompt contracts, progress logs, role and employee context, review/report metadata, verification fields, delivery package, self-evolution fields | Scheduling, board projection, collaboration, review/report, and user-visible progress reads should use `DelegationWorkItem.metadata`. | +| `work_item` | `work_kind`, `current_turn_mode`, dependency/waiting ids, handoff/context preview, prompt contracts, `progress_log`, role and employee context, review/report metadata, verification fields, delivery package, self-evolution fields | Scheduling, board projection, collaboration, review/report, and user-visible progress reads should use `DelegationWorkItem.metadata`. | | `runtime_task` | `runtime_v2`, `runtime_verification*`, `member_session_state`, `external_resume_*`, `working_memory`, `interrupted_recovery`, `last_stop_reason`, `peer_wait`, comms reactivation audit, `runtime_control_state`, runtime session team/seat ids | Execution infrastructure should store these only on runtime `Task.metadata`. They must not become WorkItem business facts. | | `execution_copy` | `mode`, `execution_mode`, `runtime_model`, `company_profile`, organization/runtime topology, delegation ids, seat/role routing, execution-agent selection, workspace/comms roots, parent session id | Runtime `Task` may carry these as immutable routing and UI envelope copies. They are not the authoritative business facts. | diff --git a/pyproject.toml b/pyproject.toml index 88123fd..96a11b0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -73,6 +73,11 @@ all = [ "matrix-nio>=0.24.0", ] +dev = [ + "pytest>=9.0", + "pytest-timeout>=2.3", +] + [project.scripts] opc = "opc.cli.app:main" @@ -93,3 +98,8 @@ include = [ "README.md", "pyproject.toml", ] + +[tool.pytest.ini_options] +# Backstop against wedged tests: a hung test fails after 5 minutes instead +# of stalling the whole suite (requires pytest-timeout, in the `dev` extra). +timeout = 300 diff --git a/tests/cli_board/test_actions.py b/tests/cli_board/test_actions.py index c16307b..f6aba52 100644 --- a/tests/cli_board/test_actions.py +++ b/tests/cli_board/test_actions.py @@ -1,6 +1,11 @@ from __future__ import annotations +import asyncio +import os +import tempfile import unittest +import uuid +from unittest import mock from unittest.mock import AsyncMock from opc.core.models import ExecutionCheckpoint, Task, TaskStatus @@ -53,17 +58,154 @@ class _StubFacade: return self._engine +class _StubServiceResult: + def __init__(self, payload: dict) -> None: + self.payload = payload + self.events: list = [] + + +class _StubSessionService: + """In-memory stand-in for the office SessionService seam. + + Mirrors the slice of the production contract that ``BoardActions`` + relies on: ``create`` persists a session-backed placeholder Task and + ensures its memory session, ``send`` routes execution to + ``engine.process_message`` with the origin task's session, and ``stop`` + cancels the target task row. Every call is recorded for assertions. + """ + + def __init__(self, engine: _StubEngine, project_id: str) -> None: + self._engine = engine + self._project_id = project_id + self.create_calls: list[dict] = [] + self.send_calls: list[dict] = [] + self.stop_calls: list[dict] = [] + + async def create(self, **kwargs) -> _StubServiceResult: + self.create_calls.append(dict(kwargs)) + project_id = kwargs.get("project_id", self._project_id) + title = str(kwargs.get("title", "") or "New Chat") + task_id = str(uuid.uuid4()) + session_id = str(uuid.uuid4()) + await self._engine.memory.ensure_session( + session_id=session_id, + project_id=project_id, + title=title, + mode="primary", + metadata={"interface": kwargs.get("interface", "office_ui")}, + ) + await self._engine.store.save_task( + Task( + id=task_id, + title=title, + description=str(kwargs.get("description", "") or ""), + project_id=project_id, + session_id=session_id, + metadata={"exec_mode": kwargs.get("exec_mode") or "task"}, + ) + ) + return _StubServiceResult({"project_id": project_id, "task_id": task_id, "session_id": session_id}) + + async def send(self, **kwargs) -> _StubServiceResult: + self.send_calls.append(dict(kwargs)) + task = await self._engine.store.get_task(str(kwargs.get("task_id", "") or "")) + if task is None: + raise ValueError(f"task_not_found: {kwargs.get('task_id')}") + response = await self._engine.process_message( + str(kwargs.get("content", "") or ""), + project_id=kwargs.get("project_id", self._project_id), + session_id=str(task.session_id or ""), + mode=kwargs.get("mode", "task"), + origin_task_id=task.id, + ) + return _StubServiceResult( + { + "project_id": kwargs.get("project_id", self._project_id), + "task_id": task.id, + "session_id": str(task.session_id or ""), + "response": response, + } + ) + + async def stop(self, **kwargs) -> _StubServiceResult: + self.stop_calls.append(dict(kwargs)) + task = await self._engine.store.get_task(str(kwargs.get("task_id", "") or "")) + if task is None: + raise ValueError(f"target_not_found: {kwargs.get('task_id')}") + task.status = TaskStatus.CANCELLED + await self._engine.store.save_task(task) + return _StubServiceResult( + { + "project_id": kwargs.get("project_id", self._project_id), + "task_id": task.id, + "status": "cancelled", + } + ) + + +class _StubOfficeServices: + def __init__(self, engine: _StubEngine, project_id: str = "demo") -> None: + self.session = _StubSessionService(engine, project_id) + + class BoardActionsTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + super().setUp() + # Safety net: even if a code path slipped past the factory patch it + # must never touch the real OPC home. + tmp = tempfile.TemporaryDirectory(prefix="opc-cli-board-actions-test-") + self.addCleanup(tmp.cleanup) + env_patcher = mock.patch.dict(os.environ, {"OPC_HOME": tmp.name}) + env_patcher.start() + self.addCleanup(env_patcher.stop) + + def _make_actions(self, engine: _StubEngine) -> tuple[BoardActions, _StubOfficeServices]: + """Patch the OfficeServiceFactory seam so BoardActions uses our stubs. + + ``BoardActions._run_office_service`` builds a real + ``OfficeServiceFactory`` (and with it a real engine + ui_state.db) per + operation; the tests replace that seam with an async context manager + yielding stub services bound to the stub engine. + """ + services = _StubOfficeServices(engine) + + class _StubFactory: + def __init__(self, **_kwargs) -> None: + pass + + async def __aenter__(self) -> _StubOfficeServices: + return services + + async def __aexit__(self, exc_type, exc, tb) -> None: + return None + + patcher = mock.patch( + "opc.plugins.cli_board.services.actions.OfficeServiceFactory", + _StubFactory, + ) + patcher.start() + self.addCleanup(patcher.stop) + return BoardActions(_StubFacade(engine), project_id="demo"), services + async def test_create_task_creates_session_backed_placeholder(self) -> None: engine = _StubEngine() - actions = BoardActions(_StubFacade(engine), project_id="demo") + actions, services = self._make_actions(engine) task = await actions.create_task(title="Draft feature", description="Initial plan") self.assertIn(task.id, engine.store.tasks) + stored = engine.store.tasks[task.id] + self.assertEqual(stored.title, "Draft feature") + self.assertEqual(stored.description, "Initial plan") + self.assertTrue(stored.session_id) engine.memory.ensure_session.assert_awaited() - self.assertEqual(engine.store.tasks[task.id].title, "Draft feature") - self.assertEqual(engine.store.tasks[task.id].metadata["source"], "cli_board") + self.assertEqual(len(services.session.create_calls), 1) + call = services.session.create_calls[0] + self.assertEqual(call["project_id"], "demo") + self.assertEqual(call["title"], "Draft feature") + self.assertEqual(call["description"], "Initial plan") + self.assertEqual(call["exec_mode"], "task") + self.assertEqual(call["interface"], "cli_board") async def test_send_session_message_routes_through_origin_task(self) -> None: engine = _StubEngine() @@ -76,19 +218,33 @@ class BoardActionsTests(unittest.IsolatedAsyncioTestCase): project_id="demo", ) await engine.store.save_task(task) - actions = BoardActions(_StubFacade(engine), project_id="demo") + actions, services = self._make_actions(engine) response = await actions.send_session_message("task-1", "please continue") self.assertEqual(response, "ok") + self.assertEqual(len(services.session.send_calls), 1) + call = services.session.send_calls[0] + self.assertEqual(call["project_id"], "demo") + self.assertEqual(call["task_id"], "task-1") + self.assertEqual(call["content"], "please continue") + self.assertEqual(call["mode"], "task") engine.process_message.assert_awaited_once() kwargs = engine.process_message.await_args.kwargs self.assertEqual(kwargs["session_id"], "session-1") self.assertEqual(kwargs["origin_task_id"], "task-1") - self.assertEqual(engine.store.tasks["task-1"].status, TaskStatus.IDLE) async def test_cancel_task_marks_related_tasks_and_checkpoints_cancelled(self) -> None: engine = _StubEngine() + started = asyncio.Event() + release = asyncio.Event() # never set; the run only ends via cancellation + + async def _blocked_process_message(*_args, **_kwargs): + started.set() + await release.wait() + return "unreachable" + + engine.process_message = AsyncMock(side_effect=_blocked_process_message) root = Task( id="root", title="Root task", @@ -116,11 +272,17 @@ class BoardActionsTests(unittest.IsolatedAsyncioTestCase): task_id="root", ) ) - actions = BoardActions(_StubFacade(engine), project_id="demo") + actions, services = self._make_actions(engine) + + background = asyncio.create_task(actions.send_session_message("root", "keep going")) + await asyncio.wait_for(started.wait(), timeout=5) await actions.cancel_task("root") + with self.assertRaises(asyncio.CancelledError): + await background self.assertEqual(engine.store.tasks["root"].status, TaskStatus.CANCELLED) self.assertEqual(engine.store.tasks["child"].status, TaskStatus.CANCELLED) self.assertEqual(engine.store.resolved, [("cp-root", "cancelled")]) - + self.assertEqual(len(services.session.stop_calls), 1) + self.assertEqual(services.session.stop_calls[0]["task_id"], "root") diff --git a/tests/cli_board/test_tui_app.py b/tests/cli_board/test_tui_app.py index 12c243b..0d8383a 100644 --- a/tests/cli_board/test_tui_app.py +++ b/tests/cli_board/test_tui_app.py @@ -87,8 +87,12 @@ class CliBoardAppPilotTests(unittest.IsolatedAsyncioTestCase): await pilot.pause() self.assertEqual(app.state.pane_focus, "context") - await pilot.press("right") + # Selecting the running task earlier auto-focused the Session + # context tab (CliBoardApp._load_selected_detail), so cycling + # right moves session -> activity. self.assertEqual(app.state.context_tab, "session") + await pilot.press("right") + self.assertEqual(app.state.context_tab, "activity") await pilot.press("f") self.assertFalse(app.state.show_done) diff --git a/tests/test_cli_app.py b/tests/test_cli_app.py index e725c36..d441664 100644 --- a/tests/test_cli_app.py +++ b/tests/test_cli_app.py @@ -4,6 +4,7 @@ import asyncio import importlib import json import os +import re import sqlite3 import tempfile import unittest @@ -58,6 +59,17 @@ from opc.plugins.office_ui.services.project import ProjectService from opc.plugins.office_ui.services.session import SessionService from opc.plugins.office_ui.services.work_item import WorkItemService +_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + + +def _plain_output(text: str) -> str: + """CLI output with ANSI styling stripped. + + Keeps plain-text assertions stable when the invoking shell forces + color (e.g. FORCE_COLOR) onto rich/typer output. + """ + return _ANSI_ESCAPE_RE.sub("", text) + class CliEscalationFormattingTests(unittest.TestCase): def setUp(self) -> None: @@ -305,7 +317,7 @@ class CliInitProjectTests(unittest.TestCase): result = runner.invoke(app, ["init", "dupe"]) self.assertNotEqual(result.exit_code, 0) - self.assertIn("Project 'dupe' already exists", result.output) + self.assertIn("Project 'dupe' already exists", _plain_output(result.output)) self.assertFalse((opc_home / "projects" / "dupe").exists()) def test_init_existing_config_cancel_preserves_config_and_does_not_create_project(self) -> None: @@ -3119,7 +3131,7 @@ class CliChannelCommandTests(unittest.TestCase): self.assertEqual(result.exit_code, 0) mock_kill.assert_called_once() - self.assertIn("PID 4321", result.stdout) + self.assertIn("PID 4321", _plain_output(result.stdout)) def test_channels_status_reports_runtime_and_capabilities(self) -> None: config = OPCConfig() @@ -3165,9 +3177,10 @@ class CliAutomationCommandTests(unittest.TestCase): result = self.runner.invoke(app, ["exec", "--help"]) self.assertEqual(result.exit_code, 0) - self.assertIn("--session-id", result.output) - self.assertIn("--resume", result.output) - self.assertIn("--stream-json", result.output) + help_text = _plain_output(result.output) + self.assertIn("--session-id", help_text) + self.assertIn("--resume", help_text) + self.assertIn("--stream-json", help_text) def test_exec_rejects_json_and_stream_json_together(self) -> None: result = self.runner.invoke(app, ["exec", "hello", "--json", "--stream-json"]) @@ -3207,9 +3220,9 @@ class CliTalentCommandTests(unittest.TestCase): self.assertEqual(import_result.exit_code, 0) self.assertEqual(hire_result.exit_code, 0) self.assertEqual(list_result.exit_code, 0) - self.assertIn("Imported 1 talent templates", import_result.stdout) - self.assertIn("Bea Backend", hire_result.stdout) - self.assertIn("engineering-backend-architect", list_result.stdout) + self.assertIn("Imported 1 talent templates", _plain_output(import_result.stdout)) + self.assertIn("Bea Backend", _plain_output(hire_result.stdout)) + self.assertIn("engineering-backend-architect", _plain_output(list_result.stdout)) config = OPCConfig.load(opc_home / "config") self.assertEqual(config.org.talent_templates, []) diff --git a/tests/test_collaboration_rpc.py b/tests/test_collaboration_rpc.py index 8abeeb4..b74ebe6 100644 --- a/tests/test_collaboration_rpc.py +++ b/tests/test_collaboration_rpc.py @@ -131,11 +131,22 @@ class CollaborationRpcTransportTests(unittest.IsolatedAsyncioTestCase): ) parser = cli_collab._build_parser() opts = parser.parse_args(["delegate_work", "--args-json-file", str(args_path)]) - with patch.object(cli_collab.os, "name", "nt"), patch.dict( + # Simulate os.name == "nt" only at the product's single decision + # point instead of patching the global os.name, which would turn + # every pathlib.Path in this process into a WindowsPath and break + # the POSIX file reads the test itself performs. The real + # rpc_env_configured() check still runs against the patched env, + # so the Windows guard branch is exercised end to end. + with patch.object( + cli_collab, + "_windows_external_rpc_env_configured", + lambda: rpc_env_configured(), + ), patch.dict( "os.environ", server.client_env, clear=True, ): + self.assertTrue(cli_collab._windows_external_rpc_env_configured()) tool_args = cli_collab._collect_tool_args(opts) payload, is_error = await cli_collab._dispatch(opts.tool, tool_args) finally: diff --git a/tests/test_company_collaboration.py b/tests/test_company_collaboration.py index a418cd0..14c05ac 100644 --- a/tests/test_company_collaboration.py +++ b/tests/test_company_collaboration.py @@ -1431,7 +1431,7 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): task_id="task-1", workspace_path="/tmp/work", run_mode="interactive", - status="failed", + status="done", metadata={"resume_session_id": "resume-token-1"}, updated_at=datetime.now(), ) @@ -1466,7 +1466,7 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): task_id="task-2", workspace_path="/tmp/work", run_mode="interactive", - status="failed", + status="done", metadata={}, updated_at=datetime.now(), ) @@ -1519,79 +1519,6 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): self.assertNotIn("external_resume_session_id", task.metadata) self.assertNotIn("external_resume_session_scope_id", task.metadata) - @unittest.skip("Filesystem handoff stack removed; see plans/task-cleanup-dead-comms.md" - ) - async def test_structured_handoff_is_persisted_and_injected(self) -> None: - with _workspace_tempdir() as tmpdir: - store = OPCStore(Path(tmpdir) / "tasks.db") - await store.initialize() - communication = CommunicationManager(store, EventBus()) - memory = DummyMemory() - - async def execute_task(task: Task) -> TaskResult: - result = TaskResult( - status=TaskStatus.DONE, - content="Decision: use SQLite\nRisk: guest posting must remain blocked", - artifacts={"workspace": "/tmp/demo", "files": ["src/app.py"]}, - ) - task.status = result.status - task.result = {"content": result.content, "artifacts": result.artifacts} - return result - - executor = CompanyWorkItemExecutor( - org_engine=DummyOrgEngine(), - communication=communication, - approval_engine=SimpleNamespace(), - memory=memory, - execute_task=execute_task, - save_task=store.save_task, - ) - - upstream = Task( - id="planning-task", - title="Planning", - project_id="proj1", - assigned_to="reviewer", - status=TaskStatus.DONE, - result={"content": "Plan approved with clear milestones.", "artifacts": {}}, - metadata={ - "work_item_projection_id": "planning", - "work_item_summary_for_downstream": "Plan approved", - "decisions": ["Use SQLite for local persistence"], - "risks": ["Guest posting must remain blocked"], - "artifacts": ["doc: docs/plan.md"], - "acceptance_criteria": ["Implementation follows approved milestones"], - }, - ) - downstream = Task( - id="execution-task", - title="Execution", - project_id="proj1", - assigned_to="executor", - status=TaskStatus.PENDING, - dependencies=["planning"], - metadata={ - "work_item_projection_id": "execution", - "work_item_role_id": "executor", - "work_item_execution_strategy": "native", - "work_item_gate": None, - "progress_log": [], - }, - ) - await store.save_task(upstream) - await store.save_task(downstream) - - await executor._run_work_item(downstream, {"planning": upstream, "execution": downstream}) - - handoffs = await store.get_handoff_records(project_id="proj1", target_projection_id="execution") - self.assertEqual(len(handoffs), 1) - self.assertEqual(handoffs[0].payload["decisions"], ["Use SQLite for local persistence"]) - self.assertIn("Objective: Planning", downstream.metadata["handoff_context"]) - self.assertIn("Use SQLite for local persistence", downstream.metadata["handoff_context"]) - self.assertEqual(downstream.context_snapshot["handoff_payloads"][0]["source_projection_id"], "planning") - self.assertEqual(memory.calls, []) - await store.close() - async def test_company_gate_prefers_structured_review_verdict_and_persists_work_item_state(self) -> None: memory = DummyMemory() @@ -7197,65 +7124,6 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): ) await store.close() - @unittest.skip("Filesystem handoff stack removed; see plans/task-cleanup-dead-comms.md" - ) - async def test_required_handoff_records_are_persisted_as_sent_and_received(self) -> None: - # The agent-facing `ack_handoff` / `review_handoff` tools were - # deleted; the underlying handoff record is still created and - # transitions sent → received on inbox read. The former - # acked/accepted transitions used to require tool calls and are - # now obsolete. - with _workspace_tempdir() as tmpdir: - store = OPCStore(Path(tmpdir) / "tasks.db") - await store.initialize() - communication = CommunicationManager(store, EventBus()) - target_task = Task( - id="review-task", - title="Review", - project_id="proj1", - assigned_to="reviewer", - metadata={ - "work_item_projection_id": "review", - "execution_mode": "company_mode", - "workspace_root": str(tmpdir), - "output_root": str(Path(tmpdir) / "deliverables"), - "target_output_dir": str(Path(tmpdir) / "deliverables"), - "comms_root": str(Path(tmpdir) / ".opc-comms"), - }, - ) - await store.save_task(target_task) - - message = await communication.send_handoff( - task_id=target_task.id, - from_agent="executor", - to_agent="reviewer", - subject="Execution handoff", - body="Please review the implementation package.", - handoff={ - "handoff_id": "handoff-1", - "summary": "Execution package ready", - "source_projection_id": "execution", - "target_projection_id": "review", - }, - requires_ack=True, - ) - sent = await store.get_handoff_record("handoff-1") - assert sent is not None - self.assertEqual(sent.status, "sent") - self.assertEqual(message.metadata["handoff_id"], "handoff-1") - - _ = await communication.read_inbox( - agent_id="reviewer", - task=target_task, - unread_only=True, - limit=10, - mark_read=True, - ) - received = await store.get_handoff_record("handoff-1") - assert received is not None - self.assertEqual(received.status, "received") - await store.close() - async def test_send_dm_writes_file_comms_and_read_inbox_projects_from_file(self) -> None: with _workspace_tempdir() as tmpdir: store = OPCStore(Path(tmpdir) / "tasks.db") @@ -8911,64 +8779,6 @@ class CompanyCollaborationTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(inbox[0]["subject"], "Cross-functional copy review") await store.close() - @unittest.skip("Filesystem handoff stack removed; see plans/task-cleanup-dead-comms.md" - ) - async def test_context_assembler_includes_ownership_contract_and_pending_handoffs(self) -> None: - class _MemoryStub: - async def build_focused_memory_context(self, **_kwargs: object) -> str: - return "" - - async def build_memory_context(self, **_kwargs: object) -> str: - return "" - - with _workspace_tempdir() as tmpdir: - store = OPCStore(Path(tmpdir) / "tasks.db") - await store.initialize() - communication = CommunicationManager(store, EventBus()) - task = Task( - id="execution-work-item", - session_id="sess-owner", - project_id="proj1", - assigned_to="executor", - metadata={ - "execution_mode": "company_mode", - "work_item_projection_id": "execution", - "work_item_projection_title": "Engineering Execution", - "work_item_turn_type": "execute", - "member_session_id": "member::proj1::executor::eng-1", - "ownership_contract": { - "summary": "Implement only the assigned API slice.", - "write_scope": str((Path(tmpdir) / "workspace").resolve()), - "expected_artifacts": ["Updated API implementation", "Verification evidence"], - "downstream_consumer": ["reviewer"], - "allowed_collaboration_targets": ["reviewer", "cto"], - }, - }, - ) - await store.save_task(task) - await communication.send_handoff( - task_id=task.id, - from_agent="planner", - to_agent="executor", - subject="Plan handoff", - body="Use the approved API contract.", - handoff={ - "handoff_id": "handoff-ctx", - "summary": "Approved API contract", - "source_projection_id": "planning", - "target_projection_id": "execution", - }, - requires_ack=True, - ) - assembler = ContextAssembler(_MemoryStub(), store=store, communication=communication) - system_context = await assembler.build_system_context(task, role_id="executor") - - self.assertIn("## Topology", system_context) - self.assertIn("Write scope:", system_context) - self.assertIn("Pending Handoff Acknowledgements", system_context) - self.assertIn("handoff-ctx", system_context) - await store.close() - async def test_context_assembler_renders_runtime_owned_mailbox_and_manager_board_summary(self) -> None: class _MemoryStub: async def build_focused_memory_context(self, **_kwargs: object) -> str: diff --git a/tests/test_company_recruiter.py b/tests/test_company_recruiter.py index 62cbdd3..d73408e 100644 --- a/tests/test_company_recruiter.py +++ b/tests/test_company_recruiter.py @@ -30,7 +30,15 @@ from opc.layer2_organization.talent_market import TalentMarket from opc.layer5_memory.memory_manager import MemoryManager from tests._temp_paths import WorkspaceTemporaryDirectory, workspace_path -tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] +_REAL_TEMPORARY_DIRECTORY = tempfile.TemporaryDirectory + + +def setUpModule() -> None: + tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] + + +def tearDownModule() -> None: + tempfile.TemporaryDirectory = _REAL_TEMPORARY_DIRECTORY # type: ignore[assignment] class DummyRecruiterLLM: diff --git a/tests/test_company_reorg.py b/tests/test_company_reorg.py index 51a3733..5c9bd59 100644 --- a/tests/test_company_reorg.py +++ b/tests/test_company_reorg.py @@ -20,7 +20,15 @@ from opc.layer2_organization.org_engine import OrgEngine from opc.layer2_organization.reorg_manager import ReorgManager from tests._temp_paths import WorkspaceTemporaryDirectory -tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] +_REAL_TEMPORARY_DIRECTORY = tempfile.TemporaryDirectory + + +def setUpModule() -> None: + tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] + + +def tearDownModule() -> None: + tempfile.TemporaryDirectory = _REAL_TEMPORARY_DIRECTORY # type: ignore[assignment] class CompanyReorgTests(unittest.IsolatedAsyncioTestCase): diff --git a/tests/test_engine_session_defaults.py b/tests/test_engine_session_defaults.py index 47d509c..13983ef 100644 --- a/tests/test_engine_session_defaults.py +++ b/tests/test_engine_session_defaults.py @@ -11,7 +11,15 @@ from opc.engine import OPCEngine from opc.layer5_memory.secretary_policy import SecretaryPolicyManager from tests._temp_paths import WorkspaceTemporaryDirectory, workspace_path -tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] +_REAL_TEMPORARY_DIRECTORY = tempfile.TemporaryDirectory + + +def setUpModule() -> None: + tempfile.TemporaryDirectory = WorkspaceTemporaryDirectory # type: ignore[assignment] + + +def tearDownModule() -> None: + tempfile.TemporaryDirectory = _REAL_TEMPORARY_DIRECTORY # type: ignore[assignment] class _StubStore: diff --git a/tests/test_external_agent_monitoring.py b/tests/test_external_agent_monitoring.py index c067d40..fcfadea 100644 --- a/tests/test_external_agent_monitoring.py +++ b/tests/test_external_agent_monitoring.py @@ -1506,7 +1506,7 @@ class ExternalAgentMonitoringTests(unittest.IsolatedAsyncioTestCase): self.assertNotIn("## Skill: memory", worker_task.description) self.assertIn("## Skill: memory", final_task.description) - async def test_engine_stages_uploaded_attachments_for_external_resume_prompt(self) -> None: + async def test_engine_provisions_uploaded_attachments_for_external_resume_prompt(self) -> None: engine = OPCEngine() with tempfile.TemporaryDirectory() as tmpdir: @@ -2800,7 +2800,6 @@ class ExternalAgentMonitoringTests(unittest.IsolatedAsyncioTestCase): proc = type("Proc", (), {"pid": 321, "stdin": None, "stdout": object(), "stderr": object()})() task = Task(title="demo", description="demo") prompt = adapter.build_task_prompt(task) - cmd, metadata = adapter.build_interactive_invocation(task, workspace_path="/repo") tmpdir = _make_test_dir("codex-no-pty-argv-prompt") try: @@ -2809,6 +2808,12 @@ class ExternalAgentMonitoringTests(unittest.IsolatedAsyncioTestCase): "opc.layer3_agent.adapters.codex_adapter.asyncio.create_subprocess_exec", AsyncMock(return_value=proc), ) as spawn_mock: + # Build inside the no-PTY patch: build_interactive_invocation + # records stdin_policy/interactive_input_channel in metadata, + # and on a real PTY-less host build and start observe the same + # platform capability. Building outside the patch would bake in + # this host's PTY support and contradict the patched start. + cmd, metadata = adapter.build_interactive_invocation(task, workspace_path="/repo") started = await adapter.start_process( cmd, tmpdir, diff --git a/tests/test_fix4_and_fix6.py b/tests/test_fix4_and_fix6.py index c453e3c..f38f1ee 100644 --- a/tests/test_fix4_and_fix6.py +++ b/tests/test_fix4_and_fix6.py @@ -18,7 +18,7 @@ import tempfile import unittest from pathlib import Path -from opc.core.models import DelegationWorkItem, Phase, RoleRuntimeSession +from opc.core.models import DelegationWorkItem, Phase, RoleRuntimeSession, Task from opc.database.store import OPCStore from opc.layer2_organization import phase_hooks # noqa: F401 (register hooks) from opc.layer3_agent.adapters.base import ExternalAgentAdapter @@ -126,7 +126,15 @@ class ReviewPromptSchemaTests(unittest.TestCase): self.assertNotIn("auto-rejected", text.lower()) def test_native_agent_prompt_keeps_suggested_schema(self) -> None: - text = native_agent._COMPANY_REVIEW_WORK_ITEM_GUIDELINES + # The guidelines live in company_runtime_contract; the native agent + # injects them through build_company_work_item_contract, so assert + # through the wiring the native agent actually uses. + task = Task( + id="review-task", + title="Review work item", + metadata={"work_item_turn_type": "review"}, + ) + text = native_agent.build_company_work_item_contract(task) self.assertIn("review_verdict", text) self.assertIn("blocking_issues", text) self.assertNotIn("Mandatory verdict schema", text) diff --git a/tests/test_metadata_ownership.py b/tests/test_metadata_ownership.py index 8690786..bf29120 100644 --- a/tests/test_metadata_ownership.py +++ b/tests/test_metadata_ownership.py @@ -432,6 +432,8 @@ class MetadataOwnershipMigratorTests(unittest.IsolatedAsyncioTestCase): "work_item_projection_id": "ceo::deliver::1", "work_item_turn_type": "deliver", "authoritative_output": True, + "user_visible": True, + "feedback_scope": "final", }, ) set_linked_work_item_id(task, item.work_item_id) diff --git a/tests/test_native_runtime_v2.py b/tests/test_native_runtime_v2.py index 4fb9526..888bb47 100644 --- a/tests/test_native_runtime_v2.py +++ b/tests/test_native_runtime_v2.py @@ -976,7 +976,7 @@ class NativeRuntimeV2Tests(unittest.IsolatedAsyncioTestCase): ), ) - # Must NOT block on a human — completes so the company workflow advances. + # Must NOT block on a human — completes so the company run advances. self.assertEqual(result.status, TaskStatus.DONE) # The failed verdict is still recorded for audit / downstream review. self.assertIn("verification", result.artifacts) diff --git a/tests/test_office_shutdown_lifecycle.py b/tests/test_office_shutdown_lifecycle.py index ae8b862..7c25ba6 100644 --- a/tests/test_office_shutdown_lifecycle.py +++ b/tests/test_office_shutdown_lifecycle.py @@ -9,6 +9,30 @@ from opc.core.models import ExecutionCheckpoint, Task from opc.plugins.office_ui.ws_handler import WSHandler +def _make_handler(root_engine: SimpleNamespace | None = None, **overrides: object) -> WSHandler: + """Build a real WSHandler over stub dependencies. + + Constructed through ``__init__`` (rather than ``__new__`` plus a + hand-copied attribute list) so every attribute ``shutdown()`` touches is + initialized by the constructor itself; each test only overrides what its + scenario controls. ``__init__`` is side-effect free for stub inputs: it + assigns state, builds Dispatcher/OfficeServices holders, and wires engine + callbacks via defensive setattr/getattr. + """ + engine = root_engine if root_engine is not None else SimpleNamespace() + if not hasattr(engine, "project_id"): + engine.project_id = "default" + handler = WSHandler( + engine=engine, + agent_store=SimpleNamespace(), + chat_store=None, + event_adapter=SimpleNamespace(), + ) + for name, value in overrides.items(): + setattr(handler, name, value) + return handler + + def test_ws_shutdown_checkpoints_before_cancelling_and_awaiting_sessions() -> None: async def scenario() -> None: events: list[str] = [] @@ -29,18 +53,12 @@ def test_ws_shutdown_checkpoints_before_cancelling_and_awaiting_sessions() -> No execution_task = asyncio.create_task(execution()) await started.wait() - handler = WSHandler.__new__(WSHandler) - handler.engine = SimpleNamespace() - handler._root_engine = SimpleNamespace( - prepare_active_company_runtimes_for_shutdown=prepare, + handler = _make_handler( + SimpleNamespace(prepare_active_company_runtimes_for_shutdown=prepare), + _background_tasks={execution_task}, + _task_bg_context={execution_task: {"task_id": "runtime-task"}}, + _task_bg_map={"runtime-task": {execution_task}}, ) - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = set() - handler._active_message_tasks = set() - handler._background_tasks = {execution_task} - handler._task_bg_context = {execution_task: {"task_id": "runtime-task"}} - handler._task_bg_map = {"runtime-task": {execution_task}} await handler.shutdown(timeout=1.0) @@ -63,20 +81,16 @@ def test_ws_shutdown_checkpoint_failure_does_not_cancel_execution_or_close_the_g execution_task = asyncio.create_task(execution()) await asyncio.sleep(0) - handler = WSHandler.__new__(WSHandler) - handler.engine = SimpleNamespace() - handler._root_engine = SimpleNamespace( - prepare_active_company_runtimes_for_shutdown=AsyncMock( - side_effect=RuntimeError("checkpoint unavailable") + handler = _make_handler( + SimpleNamespace( + prepare_active_company_runtimes_for_shutdown=AsyncMock( + side_effect=RuntimeError("checkpoint unavailable") + ), ), + _background_tasks={execution_task}, + _task_bg_context={execution_task: {"task_id": "runtime-task"}}, + _task_bg_map={"runtime-task": {execution_task}}, ) - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = set() - handler._active_message_tasks = set() - handler._background_tasks = {execution_task} - handler._task_bg_context = {execution_task: {"task_id": "runtime-task"}} - handler._task_bg_map = {"runtime-task": {execution_task}} try: await handler.shutdown(timeout=1.0) @@ -101,9 +115,7 @@ def test_ws_shutdown_rejects_background_work_scheduled_by_late_ingress() -> None nonlocal entered entered = True - handler = WSHandler.__new__(WSHandler) - handler._shutting_down = True - handler._background_tasks = set() + handler = _make_handler(_shutting_down=True) task = handler._track(late_work()) await asyncio.gather(task, return_exceptions=True) await asyncio.sleep(0) @@ -133,17 +145,7 @@ def test_ws_shutdown_drains_queued_duplicate_handoff_before_checkpointing() -> N _active_task_run_registry=registry, prepare_active_company_runtimes_for_shutdown=prepare, ) - handler = WSHandler.__new__(WSHandler) - handler.engine = root_engine - handler._root_engine = root_engine - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = set() - handler._active_message_tasks = set() - handler._background_tasks = set() - handler._task_bg_context = {} - handler._task_bg_map = {} - handler._handoff_route_tasks = {} + handler = _make_handler(root_engine) async def execution() -> None: async with runtime_lock: @@ -213,20 +215,16 @@ def test_ws_shutdown_fails_closed_while_execution_cleanup_is_still_running() -> execution_task = asyncio.create_task(execution()) await asyncio.sleep(0) - handler = WSHandler.__new__(WSHandler) - handler.engine = SimpleNamespace() - handler._root_engine = SimpleNamespace( - prepare_active_company_runtimes_for_shutdown=AsyncMock(return_value=[]), + handler = _make_handler( + SimpleNamespace( + prepare_active_company_runtimes_for_shutdown=AsyncMock(return_value=[]), + ), + _background_tasks={execution_task}, + _task_bg_context={ + execution_task: {"task_id": "runtime-task", "execution_handoff": True} + }, + _task_bg_map={"runtime-task": {execution_task}}, ) - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = set() - handler._active_message_tasks = set() - handler._background_tasks = {execution_task} - handler._task_bg_context = { - execution_task: {"task_id": "runtime-task", "execution_handoff": True} - } - handler._task_bg_map = {"runtime-task": {execution_task}} try: await handler.shutdown(timeout=0.01) @@ -263,21 +261,21 @@ def test_ws_shutdown_cancels_execution_before_waiting_for_client_close() -> None execution_task = asyncio.create_task(execution()) await asyncio.sleep(0) client = BlockingWebSocket() - handler = WSHandler.__new__(WSHandler) - handler.engine = SimpleNamespace() - handler._root_engine = SimpleNamespace( - prepare_active_company_runtimes_for_shutdown=AsyncMock(return_value=[]), + handler = _make_handler( + SimpleNamespace( + prepare_active_company_runtimes_for_shutdown=AsyncMock(return_value=[]), + ), + _clients={client}, + _background_tasks={execution_task}, + _task_bg_context={execution_task: {"task_id": "runtime-task"}}, + _task_bg_map={"runtime-task": {execution_task}}, ) - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = {client} - handler._active_message_tasks = set() - handler._background_tasks = {execution_task} - handler._task_bg_context = {execution_task: {"task_id": "runtime-task"}} - handler._task_bg_map = {"runtime-task": {execution_task}} shutdown_task = asyncio.create_task(handler.shutdown(timeout=1.0)) - await close_entered.wait() + # close_entered only fires from inside shutdown's client-close loop. + # Bound the wait so a shutdown crash before that loop fails the test + # instead of wedging the whole suite. + await asyncio.wait_for(close_entered.wait(), timeout=5.0) assert execution_released.is_set() assert execution_task.done() @@ -302,19 +300,7 @@ def test_duplicate_resume_does_not_leave_shutdown_handoff_barrier_queued() -> No _active_task_run_registry=registry, prepare_active_company_runtimes_for_shutdown=prepare, ) - handler = WSHandler.__new__(WSHandler) - handler.engine = root_engine - handler._root_engine = root_engine - handler.chat_store = None - handler._shutting_down = False - handler._progress_flush_task = None - handler._clients = set() - handler._active_message_tasks = set() - handler._background_tasks = set() - handler._task_bg_context = {} - handler._task_bg_map = {} - handler._company_stop_finalize_tasks = {} - handler._company_suspend_reply_locks = {} + handler = _make_handler(root_engine) checkpoint = ExecutionCheckpoint( checkpoint_id="checkpoint-1", diff --git a/tests/test_org_concurrency.py b/tests/test_org_concurrency.py deleted file mode 100644 index b75bee6..0000000 --- a/tests/test_org_concurrency.py +++ /dev/null @@ -1,106 +0,0 @@ -"""Concurrency tests for the _config_lock pattern used in P2. - -The real WSHandler requires a full engine/store/ws stack to boot, which makes -a pure-unit concurrency test prohibitively heavy. Instead we test the *pattern* -applied in P2 — a single asyncio.Lock serializing any number of mixed -`add_role` and `reorg_decide` style coroutines — and assert: - - - Every operation completes (20 ops in our case). - - `asyncio.Lock` never produces a RuntimeError ("lock already held") — which - would indicate same-task reentry. - - The log shows strictly interleaved entries (no two tasks recorded an - "inside-lock" line between each other's enter/exit). - - The whole batch completes well under the 10s budget. - -Evidence: ws_handler.py:175 defines `self._config_lock = asyncio.Lock()`. -P2 wraps `_handle_reorg_decide`, `_handle_set_mode`, and the `company_profile` -swap under this same lock. The proof of no deadlock is static (D4 in the plan); -this test is a dynamic smoke that the cooperative-locking behavior holds. -""" -from __future__ import annotations - -import asyncio -import time - -import pytest - - -@pytest.mark.asyncio -async def test_add_role_and_reorg_decide_serialized(): - """20 mixed ops share one lock; no exceptions, no deadlock, finishes fast.""" - lock = asyncio.Lock() - # Event log records enter/exit around the critical section so we can verify - # the lock serialized the body. If two tasks ever interleaved between an - # enter and its matching exit, the test fails. - log: list[tuple[str, str]] = [] # (op, phase) - - async def guarded(op: str) -> None: - async with lock: - log.append((op, "enter")) - # Yield to event loop so other tasks get a chance to observe - # whether they see an "inside-lock" window (they should not). - await asyncio.sleep(0) - log.append((op, "exit")) - - started = time.monotonic() - tasks = [] - for i in range(10): - tasks.append(asyncio.create_task(guarded(f"add_role:{i}"))) - tasks.append(asyncio.create_task(guarded(f"reorg_decide:{i}"))) - - results = await asyncio.gather(*tasks, return_exceptions=True) - elapsed = time.monotonic() - started - - exceptions = [r for r in results if isinstance(r, BaseException)] - assert exceptions == [], f"Unexpected exceptions: {exceptions}" - - # Every op must produce exactly one enter + one exit. - assert len(log) == 40, f"Expected 40 log entries (20 ops × 2), got {len(log)}" - - # Verify lock-mutual-exclusion: iterating the log, each enter must be - # immediately followed by the SAME op's exit (no interleave). - i = 0 - while i < len(log): - assert log[i][1] == "enter", f"Expected enter at index {i}, got {log[i]}" - assert log[i + 1] == (log[i][0], "exit"), ( - f"Enter for {log[i][0]} was not immediately followed by its exit; " - f"saw {log[i + 1]} — indicates lock was not held across the body." - ) - i += 2 - - # Counts must balance. - add_role_count = sum(1 for (op, phase) in log if phase == "enter" and op.startswith("add_role")) - reorg_count = sum(1 for (op, phase) in log if phase == "enter" and op.startswith("reorg_decide")) - assert add_role_count == 10 - assert reorg_count == 10 - - # 10s budget; a correct cooperative-lock implementation finishes in <1s. - assert elapsed < 10.0, f"Test took {elapsed:.2f}s, exceeded 10s budget" - - -@pytest.mark.asyncio -async def test_lock_is_not_reentrant_from_same_task(): - """asyncio.Lock must NOT be re-acquirable from the same task (would deadlock). - - This is the invariant that the P2 plan relied on (D4 deadlock proof): if any - code path under `_handle_reorg_decide` re-acquired `_config_lock` we'd - deadlock. Here we verify the Lock primitive's behavior — the test passes - only when a re-acquire attempt blocks forever (we force it to time out). - """ - lock = asyncio.Lock() - - async def try_reentrant(): - async with lock: - # Attempting to acquire again from the same task must block. - try: - await asyncio.wait_for(lock.acquire(), timeout=0.2) - except asyncio.TimeoutError: - return "blocked_as_expected" - # Only reach here if stdlib ever became reentrant (it is not). - lock.release() - return "unexpected_reentry" - - outcome = await try_reentrant() - assert outcome == "blocked_as_expected", ( - f"Expected asyncio.Lock to block on same-task re-acquire, got {outcome}" - ) diff --git a/tests/test_parallel_runtime_isolation.py b/tests/test_parallel_runtime_isolation.py index 7db9f64..db3d74b 100644 --- a/tests/test_parallel_runtime_isolation.py +++ b/tests/test_parallel_runtime_isolation.py @@ -377,7 +377,8 @@ async def test_ui_project_switch_uses_delegate_without_cancelling_background_tas await asyncio.sleep(0) root._get_project_delegate.assert_awaited_once_with("project-b") - assert handler.engine is root + # Project switch deliberately rebinds the handler to the delegate. + assert handler.engine is delegate assert handler._client_project_ids[ws] == "project-b" assert handler._client_switch_seq[ws] == "seq-1" sent_types = [call.args[0]["type"] for call in ws.send_json.await_args_list] @@ -748,6 +749,7 @@ async def test_kanban_create_task_routes_by_request_project_id() -> None: engine_b = _ui_engine("project-b", _MemoryStore([])) chat_store = _ui_chat_store() handler = WSHandler(engine_a, MagicMock(), chat_store, _ui_event_adapter()) + engine_a._get_project_delegate = AsyncMock(return_value=engine_b) handler._engine_for_project = AsyncMock( side_effect=lambda project_id: engine_b if project_id == "project-b" else engine_a, ) @@ -877,7 +879,6 @@ def test_explicit_agent_overrides_company_role_agent_defaults() -> None: runtime_topology=topology, decision=decision, project_id="proj", - role_agent_overrides={"ceo": "codex"}, ) seat = enriched["seats"][0] diff --git a/tests/test_role_update_handler.py b/tests/test_role_update_handler.py index fd113ab..461226e 100644 --- a/tests/test_role_update_handler.py +++ b/tests/test_role_update_handler.py @@ -2,51 +2,82 @@ from __future__ import annotations import unittest from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import MagicMock, patch from opc.core.config import OPCConfig, RoleConfig -class UpdateRoleHandlerTests(unittest.IsolatedAsyncioTestCase): +def _make_role() -> RoleConfig: + return RoleConfig( + id="student", + name="Student", + responsibility="Learn.", + reports_to="owner", + tools=["file_read"], + ) + + +def _make_context(cfg: OPCConfig, *, exec_mode: str, company_profile: str): + from opc.plugins.office_ui.services.context import ModeState, OfficeServiceContext + + engine = SimpleNamespace(config=cfg, org_engine=MagicMock()) + context = OfficeServiceContext( + engine=engine, + agent_store=None, + chat_store=MagicMock(), + event_adapter=MagicMock(), + mode_state=ModeState(exec_mode=exec_mode, company_profile=company_profile), + ) + return engine, context + + +class UpdateRolePersistenceTests(unittest.IsolatedAsyncioTestCase): async def test_update_role_persists_tools(self) -> None: - from opc.plugins.office_ui.ws_handler import WSHandler + """Role tool edits persist when the active org is an editable custom org. + + Role mutation now lives in OrgService (the WS handler delegates to it), + and writes are intentionally refused for built-in read-only orgs, so + persistence must be asserted against an editable custom org. + """ + from opc.plugins.office_ui.services.org import OrgService cfg = OPCConfig() - cfg.org.roles = [ - RoleConfig( - id="student", - name="Student", - responsibility="Learn.", - reports_to="owner", - tools=["file_read"], - ) - ] + cfg.org.company_profile = "custom" + cfg.org.organization_id = "lab_org" + cfg.org.organization_name = "Lab Org" + cfg.org.roles = [_make_role()] + engine, context = _make_context(cfg, exec_mode="custom", company_profile="custom") - handler = WSHandler.__new__(WSHandler) - handler.engine = SimpleNamespace(config=cfg, org_engine=MagicMock()) - handler._clients = set() - handler._shutting_down = False - handler._ws_is_open = lambda _ws: True - handler._config_lock = AsyncMock() - handler._config_lock.__aenter__ = AsyncMock(return_value=None) - handler._config_lock.__aexit__ = AsyncMock(return_value=None) - handler._broadcast_org_info = AsyncMock() - - ws = AsyncMock() with patch.object(OPCConfig, "save", autospec=True) as save: - await handler._handle_update_role( - ws, - { - "role_id": "student", - "tools": ["file_read", " ", "web_search", ""], - }, + result = await OrgService(context).update_role( + "student", + {"tools": ["file_read", " ", "web_search", ""]}, ) self.assertEqual(cfg.org.roles[0].tools, ["file_read", "web_search"]) - handler.engine.org_engine.reload_from_config.assert_called_once() save.assert_called_once_with(cfg) - handler._broadcast_org_info.assert_awaited_once() - ws.send_json.assert_awaited() - payload = ws.send_json.call_args.args[0] - self.assertEqual(payload["type"], "ack") - self.assertTrue(payload["payload"]["ok"]) + engine.org_engine.reload_from_config.assert_called_once() + self.assertEqual(result.payload["action"], "role_updated") + self.assertEqual(result.payload["role_id"], "student") + self.assertEqual(result.payload["role"]["tools"], ["file_read", "web_search"]) + + async def test_update_role_tools_rejected_for_readonly_builtin_org(self) -> None: + """Built-in (corporate) orgs are read-only: tool edits must not persist.""" + from opc.plugins.office_ui.services.models import ServiceError + from opc.plugins.office_ui.services.org import OrgService + + cfg = OPCConfig() + cfg.org.roles = [_make_role()] + engine, context = _make_context(cfg, exec_mode="company", company_profile="corporate") + + with patch.object(OPCConfig, "save", autospec=True) as save: + with self.assertRaises(ServiceError) as raised: + await OrgService(context).update_role( + "student", + {"tools": ["file_read", "web_search"]}, + ) + + self.assertEqual(raised.exception.code, "org_read_only") + self.assertEqual(cfg.org.roles[0].tools, ["file_read"]) + save.assert_not_called() + engine.org_engine.reload_from_config.assert_not_called() diff --git a/tests/test_session_context_compression.py b/tests/test_session_context_compression.py index 4afe449..348c803 100644 --- a/tests/test_session_context_compression.py +++ b/tests/test_session_context_compression.py @@ -202,14 +202,13 @@ class SessionContextCompressionTests(unittest.IsolatedAsyncioTestCase): store = OPCStore(db_path) await store.initialize() memory = MemoryManager(root, "proj1", store=store) - memory.set_history_compactor( - HistoryCompactor( - llm=_StubLLM(), - store=store, - memory_manager=memory, - compression_threshold=0.85, - ) + compactor = HistoryCompactor( + llm=_StubLLM(), + store=store, + memory_manager=memory, + compression_threshold=0.85, ) + memory.set_history_compactor(compactor) session_id = "session-restart" for idx in range(6): @@ -217,6 +216,20 @@ class SessionContextCompressionTests(unittest.IsolatedAsyncioTestCase): payload = f"message {idx} " + ("alpha beta gamma delta " * 40) await memory.append_session_message(session_id=session_id, role=role, text=payload) + # Append no longer auto-compacts; compaction is an explicit call. + # Forced compaction consumes everything up to the latest message, + # so the raw tail comes from messages appended afterwards. + compacted = await compactor.maybe_compact_session( + project_id="proj1", + session_id=session_id, + force=True, + ) + self.assertTrue(compacted) + for idx in range(6, 8): + role = "user" if idx % 2 == 0 else "assistant" + payload = f"message {idx} " + ("alpha beta gamma delta " * 40) + await memory.append_session_message(session_id=session_id, role=role, text=payload) + snapshot = await store.get_latest_session_memory_snapshot(session_id) self.assertIsNotNone(snapshot) assert snapshot is not None @@ -225,7 +238,7 @@ class SessionContextCompressionTests(unittest.IsolatedAsyncioTestCase): context = await memory.build_session_prompt_context(session_id) self.assertIn("## Session Memory", context) self.assertIn("Session history summary before restart", context) - self.assertIn("message 5", context) + self.assertIn("message 7", context) await store.close() @@ -236,7 +249,7 @@ class SessionContextCompressionTests(unittest.IsolatedAsyncioTestCase): restarted_context = await restarted_memory.build_session_prompt_context(session_id) self.assertTrue(any("Session history summary before restart" in item["content"] for item in restarted_history)) - self.assertTrue(any("message 5" in item["content"] for item in restarted_history)) + self.assertTrue(any("message 7" in item["content"] for item in restarted_history)) self.assertFalse(any("message 0" in item["content"] for item in restarted_history)) self.assertNotIn("message 0", restarted_context) diff --git a/tests/test_work_item_runtime_links.py b/tests/test_work_item_runtime_links.py index 3d78c73..56d747e 100644 --- a/tests/test_work_item_runtime_links.py +++ b/tests/test_work_item_runtime_links.py @@ -328,7 +328,7 @@ class WorkItemRuntimeLinkTests(unittest.IsolatedAsyncioTestCase): {item.work_item_id: canonical.id}, ) - async def test_task_delete_cascades_link_and_allows_rematerialization(self) -> None: + async def test_task_delete_cascades_link_and_work_item_row(self) -> None: item = _work_item("wi-cascade") task = _task("task-cascade") await self.store.save_delegation_work_item(item) @@ -337,10 +337,13 @@ class WorkItemRuntimeLinkTests(unittest.IsolatedAsyncioTestCase): await self.store.hard_delete_task(task.id) + # Hard delete removes the linked work item row together with the + # runtime traces, so the identity cannot be rematerialized. self.assertEqual(await self.store.get_runtime_links_for_work_items([item.work_item_id]), {}) + self.assertIsNone(await self.store.get_delegation_work_item(item.work_item_id)) replacement = _task("task-cascade-replacement") await self.store.save_task(replacement) - self.assertTrue(await self.store.link_work_item_runtime_task(item.work_item_id, replacement.id)) + self.assertFalse(await self.store.link_work_item_runtime_task(item.work_item_id, replacement.id)) async def test_transition_from_task_uses_structured_link_without_legacy_metadata(self) -> None: item = _work_item("wi-transition")