"""Tests for the prompt-assembly deduplication refactor. These tests pin down the structural decisions made in the refactor: 1. Work-item identity (the global_intent_summary / your_responsibility / inputs / deliverables / acceptance_criteria / out_of_scope block) is rendered exactly ONCE in any per-turn prompt for a company-mode role. The runtime plan and ownership contract no longer re-render those fields. 2. Static collaboration boilerplate (Company Work Item Contract, the Work Item Turn: ... headers, the Blocking collaboration (rare) block, the "When to send / When to reply" prose previously baked into the comms inbox section) no longer appears in any ContextAssembler output. It now lives in the ``collaboration-playbook`` skill, injected per-turn through the skills channel for company_mode runs only. 3. ``SkillLibrary.build_skills_summary`` honors the per-skill ``modes`` frontmatter field: a skill restricted to company_mode is fully hidden — body AND description — when the call is made without a matching ``execution_mode``. 4. ``render_inbox_section`` no longer emits the "When to send" / "When to reply" prose. ``render_meetings_section`` returns "" when no meetings are open instead of emitting "(no open meetings)". The first two tests are the load-bearing ones: if a future refactor introduces a new builder that re-renders work-item identity or re-introduces a static collaboration paragraph into the per-turn prompt, these tests should fail and tell the author to either move the content into the playbook skill or pick a single canonical section to own it. """ from __future__ import annotations import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock from opc.core.models import SessionMessageRecord, SessionPartRecord from opc.core.models import Task from opc.layer1_perception.context_assembler import ContextAssembler from opc.layer5_memory.memory_manager import MemoryManager from opc.layer2_organization import comms as _comms from opc.layer2_organization.collaboration_policy import render_ownership_contract from opc.layer5_memory.skill_library import Skill, SkillLibrary # ---------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------- _STATIC_PHRASES_THAT_BELONG_IN_THE_PLAYBOOK = ( # Removed from context_assembler._build_work_item_collaboration_guidelines "## Company Work Item Contract", "## Work Item Turn: Plan / Intake / Dispatch", "## Work Item Turn: Execute", "## Work Item Turn: Review", "## Work Item Turn: Aggregate / Deliver", # Removed from build_external_runtime_mailbox_context "## Blocking collaboration (rare)", # Removed from comms.render_inbox_section "### When to send a message", "### When to reply to a message you received", # Removed from comms.render_meetings_section empty case "(no open meetings)", ) def _work_item_assignment_task(*, role: str = "ceo") -> Task: """Build a task with a representative work_item_assignment payload.""" return Task( id=f"task-{role}", title="CEO Intake", description="legacy description", assigned_to=role, metadata={ "execution_mode": "company_mode", "work_item_projection_title": "CEO Intake", "work_item_turn_type": "intake", "work_item_assignment": { "global_intent_summary": "Ship the migration to the new auth provider this quarter.", "your_responsibility": "Frame the company-wide intake brief and route slices to the C-suite.", "out_of_scope": [ "Do not implement code in this work item.", "Do not validate gate evidence in this work item.", ], "inputs": [ "Use the original user request as the mission baseline.", ], "deliverables": [ "deliverables/ceo_intake.md with Mission Summary, Scope, Risks, Routing, Acceptance", ], "acceptance_criteria": [ "Brief covers every required section and stays inside the work-item boundary.", ], }, "work_item_runtime_plan": { "turn_type": "intake", "summary": "Frame the company mission and route slices.", # The next four are intentionally duplicates of the # work_item_assignment fields above. After the refactor # they MUST NOT be rendered by _render_work_item_runtime_plan. "inputs": ["Use the original user request as the mission baseline."], "deliverables": ["deliverables/ceo_intake.md ..."], "acceptance_criteria": ["Brief covers every required section ..."], "out_of_scope": ["Do not implement code in this work item."], # Work-item-unique field that the runtime plan IS allowed # to carry. Used to verify the runtime plan still # surfaces non-duplicate content. "collaboration_expectations": [ "Direct messages for clarifications; meetings only for cross-role decisions." ], "verification_required": False, }, "ownership_contract": { # The summary and expected_artifacts fields are # intentionally duplicates of the work_item_assignment # responsibility/deliverables. After the refactor # render_ownership_contract MUST NOT render them. "summary": "Frame the company-wide intake brief and route slices to the C-suite.", "expected_artifacts": ["deliverables/ceo_intake.md ..."], # The boundary fields below SHOULD still be rendered. "write_scope": "/tmp/ceo-intake-workspace", "allowed_collaboration_targets": ["cto"], "downstream_consumer": ["cto"], }, }, ) # ---------------------------------------------------------------------- # Bucket 1: work-item identity is rendered exactly once # ---------------------------------------------------------------------- class WorkItemIdentityDeduplicationTests(unittest.TestCase): def test_work_item_runtime_plan_does_not_re_render_work_item_assignment_fields(self) -> None: assembler = ContextAssembler(memory=SimpleNamespace()) task = _work_item_assignment_task() rendered = assembler._render_work_item_runtime_plan(task.metadata["work_item_runtime_plan"]) # Headings that used to come out of work_item_runtime_plan are # now ONLY produced by company_mode._build_work_item_description / # ContextAssembler.build_task_brief. The runtime plan # renderer must not re-emit them. self.assertNotIn("Inputs:", rendered) self.assertNotIn("Deliverables:", rendered) self.assertNotIn("Acceptance Criteria:", rendered) self.assertNotIn("Out of Scope:", rendered) # Work-item-unique fields are still rendered. self.assertIn("Collaboration Expectations:", rendered) self.assertIn("Direct messages for clarifications", rendered) self.assertIn("Work item turn type: intake", rendered) self.assertIn("No automatic verification pass is required", rendered) def test_render_ownership_contract_omits_summary_and_expected_artifacts(self) -> None: task = _work_item_assignment_task() rendered = render_ownership_contract(task) # The boundary header still appears... self.assertIn("## Ownership Contract", rendered) self.assertIn("Write scope: /tmp/ceo-intake-workspace", rendered) self.assertIn("Allowed collaboration targets:", rendered) self.assertIn("Downstream consumers:", rendered) # ...but the summary and expected_artifacts strings (which # are duplicates of work-item identity) MUST NOT. self.assertNotIn("Frame the company-wide intake brief", rendered) self.assertNotIn("Expected artifacts:", rendered) self.assertNotIn("deliverables/ceo_intake.md ...", rendered) def test_build_task_brief_renders_projection_identity_once(self) -> None: assembler = ContextAssembler(memory=SimpleNamespace()) task = _work_item_assignment_task() brief = assembler.build_task_brief(task) # Work-item identity headers appear exactly once. self.assertEqual(brief.count("## Global Intent Summary"), 1) self.assertEqual(brief.count("## Your Responsibility"), 1) self.assertEqual(brief.count("## Inputs"), 1) self.assertEqual(brief.count("## Deliverables"), 1) self.assertEqual(brief.count("## Acceptance Criteria"), 1) self.assertEqual(brief.count("## Out of Scope"), 1) # The literal responsibility text from work_item_assignment also # appears exactly once across the entire brief. self.assertEqual( brief.count("Frame the company-wide intake brief and route slices to the C-suite."), 1, ) # runtime role map is no longer included in build_task_brief # — it was superseded by the unified Topology section in the # collaboration context path. Pinning this down so a future # refactor doesn't accidentally re-introduce it. self.assertNotIn("## Runtime Role Map", brief) # Ownership contract is also no longer included in # build_task_brief — it lives only in the collaboration # path now, so the brief and the system context don't # double-render it for native agents. self.assertNotIn("## Ownership Contract", brief) # ---------------------------------------------------------------------- # Bootstrap work-item assignment input/acceptance deduplication # ---------------------------------------------------------------------- class RuntimeWorkItemAcceptanceDedupTests(unittest.TestCase): """Pin down generic acceptance criteria deduplication.""" def _planner(self) -> Any: from opc.layer2_organization.company_mode import CompanyRuntimeSpecBuilder class _OrgEngine: def get_agent(self, role_id: str): return None return CompanyRuntimeSpecBuilder(_OrgEngine()) def _projection_spec(self, *, dependencies: list[str]) -> Any: from opc.layer2_organization.org_work_item_planner import WorkItemProjectionSpec return WorkItemProjectionSpec( projection_id="cto_synthesis", turn_type="aggregate", title="CTO Synthesis", summary="Review child outputs and synthesize the engineering handoff.", role_id="cto", dependency_projection_ids=list(dependencies), execution_strategy="auto", ) def test_generic_acceptance_criteria_omit_main_deliverable_echo(self) -> None: planner = self._planner() projection_spec = self._projection_spec(dependencies=["senior_engineer__execute"]) deliverables = [ "A concrete output that fulfills this work item objective and can be handed off downstream.", ] criteria = planner._infer_work_item_acceptance(projection_spec, deliverables) self.assertFalse( any(item.startswith("The main deliverable is present:") for item in criteria), msg=f"Generic acceptance criteria still echo the first deliverable verbatim: {criteria}", ) self.assertNotIn(deliverables[0], "\n".join(criteria)) # ---------------------------------------------------------------------- # Bucket 2: Self section consolidation # ---------------------------------------------------------------------- class SelfSectionConsolidationTests(unittest.IsolatedAsyncioTestCase): async def test_self_section_merges_identity_persona_and_delta(self) -> None: memory = SimpleNamespace( build_agent_memory_context=AsyncMock(return_value=""), build_memory_context=AsyncMock(return_value=""), ) assembler = ContextAssembler(memory=memory, store=SimpleNamespace()) task = Task( id="self-section-task", title="Engineering", assigned_to="senior_engineer", metadata={ "execution_mode": "company_mode", "employee_assignment": { "name": "Backend Architect", "employee_id": "backend-architect", "role_id": "senior_engineer", "category": "engineering", "domains": ["coding", "api"], "experience_score": 5, }, "employee_prompt_context": "Backend specialist focused on APIs.", "employee_delta_context": "## Working Patterns\n- Leave reviewer-friendly artifacts.", }, ) rendered = await assembler.build_role_reference_context(task, role_id="senior_engineer") # Single canonical Self section with role, employee, and profile sub-blocks. self.assertIn("## Self", rendered) self.assertIn("### Role", rendered) self.assertIn("- Role: senior_engineer", rendered) self.assertIn("### Employee", rendered) self.assertIn("- Employee: Backend Architect", rendered) self.assertIn("### Employee Persona", rendered) self.assertIn("Backend specialist focused on APIs.", rendered) self.assertIn("### Learned Working Profile", rendered) # The legacy independent headers must NOT be present. self.assertNotIn("## Assigned Employee", rendered) self.assertNotRegex(rendered, r"(?m)^## Employee Persona$") self.assertNotIn("## Employee Delta Profile", rendered) # ---------------------------------------------------------------------- # Static collaboration boilerplate is gone from per-turn prompt # ---------------------------------------------------------------------- class StaticCollaborationBoilerplateRemovedTests(unittest.IsolatedAsyncioTestCase): async def test_no_static_collaboration_phrases_in_external_context(self) -> None: memory = SimpleNamespace( build_focused_memory_context=AsyncMock(return_value=""), build_memory_context=AsyncMock(return_value=""), build_agent_memory_context=AsyncMock(return_value=""), ) # We do not need a real CommunicationManager — just enough # surface for build_collaboration_context to short-circuit. communication = SimpleNamespace( build_agent_context=AsyncMock( return_value={ "inbox": [], "annotations": [], "pending_handoffs": [], "allowed_contacts": [ { "role_id": "cto", "name": "CTO", "relation": "downstream", "responsibility": "Owns technical execution", } ], } ) ) assembler = ContextAssembler(memory=memory, store=None, communication=communication) task = _work_item_assignment_task() external_ctx = await assembler.build_external_context(task, role_id="ceo") for phrase in _STATIC_PHRASES_THAT_BELONG_IN_THE_PLAYBOOK: self.assertNotIn( phrase, external_ctx, msg=f"Static phrase {phrase!r} should now live in the collaboration-playbook skill, not in per-turn context", ) async def test_task_mode_external_context_omits_collaboration_mailbox_and_comms_prompting(self) -> None: memory = SimpleNamespace( build_focused_memory_context=AsyncMock(return_value=""), build_memory_context=AsyncMock(return_value=""), build_agent_memory_context=AsyncMock(return_value=""), ) assembler = ContextAssembler( memory=memory, store=None, communication=SimpleNamespace(build_agent_context=AsyncMock(return_value={})), ) task = Task( id="task-mode-external", title="Task Mode External", description="Handle the task directly.", assigned_to="task_generalist", metadata={ "mode": "task", "execution_mode": "task_mode", "work_item_role_id": "task_generalist", }, context_snapshot={}, ) external_ctx = await assembler.build_external_context(task, role_id="task_generalist") self.assertNotIn("## Continue After Blocking Collaboration", external_ctx) self.assertNotIn("## External Collaboration Actions", external_ctx) self.assertNotIn("## Comms", external_ctx) def test_render_inbox_section_omits_when_to_send_prose(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: layout = _comms.resolve_layout(tmpdir, "proj1", "sess1") _comms.ensure_layout(layout, ["ceo", "cto"]) rendered = _comms.render_inbox_section(layout, "ceo") self.assertIn("### Mailbox", rendered) self.assertIn("runtime-owned", rendered) # The standing rules now live in the playbook skill. self.assertNotIn("### When to send a message", rendered) self.assertNotIn("### When to reply to a message you received", rendered) self.assertNotIn("Send ONLY when", rendered) self.assertNotIn("Reply ONLY when", rendered) self.assertNotIn("call `read_inbox`", rendered) def test_render_meetings_section_returns_empty_when_no_meetings(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: layout = _comms.resolve_layout(tmpdir, "proj1", "sess1") _comms.ensure_layout(layout, ["ceo"]) rendered = _comms.render_meetings_section(layout, "ceo") self.assertEqual(rendered, "") # ---------------------------------------------------------------------- # Current-turn replay is omitted from prompt history # ---------------------------------------------------------------------- class PromptHistoryReplayTests(unittest.IsolatedAsyncioTestCase): async def test_session_prompt_builders_skip_latest_user_turn_when_requested(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: memory = MemoryManager(Path(tmpdir)) earlier_user = SessionMessageRecord(session_id="sess-1", role="user") assistant = SessionMessageRecord(session_id="sess-1", role="assistant") current_user = SessionMessageRecord(session_id="sess-1", role="user") visible_items = [ { "message": earlier_user, "parts": [SessionPartRecord(message_id=earlier_user.message_id, session_id="sess-1", payload={"text": "first request"})], }, { "message": assistant, "parts": [SessionPartRecord(message_id=assistant.message_id, session_id="sess-1", payload={"text": "assistant reply"})], }, { "message": current_user, "parts": [SessionPartRecord(message_id=current_user.message_id, session_id="sess-1", payload={"text": "current request"})], }, ] memory._get_visible_session_transcript = AsyncMock(return_value=visible_items) # type: ignore[method-assign] memory.build_session_memory_context = AsyncMock(return_value="") # type: ignore[method-assign] full_history = await memory.build_session_history_messages("sess-1") trimmed_history = await memory.build_session_history_tail_messages( "sess-1", include_latest_user_turn=False, ) prompt_context = await memory.build_session_prompt_context( "sess-1", include_latest_user_turn=False, ) self.assertEqual(len(full_history), 3) self.assertEqual(len(trimmed_history), 2) self.assertIn("first request", trimmed_history[0]["content"]) self.assertIn("assistant reply", trimmed_history[1]["content"]) self.assertNotIn("current request", prompt_context) self.assertIn("first request", prompt_context) self.assertIn("assistant reply", prompt_context) async def test_child_session_seed_message_is_omitted_from_prompt_history(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: memory = MemoryManager(Path(tmpdir)) seed_user = SessionMessageRecord( session_id="sess-2", role="user", metadata={"kind": "child_session_seed"}, ) assistant = SessionMessageRecord(session_id="sess-2", role="assistant") followup_user = SessionMessageRecord(session_id="sess-2", role="user") visible_items = [ { "message": seed_user, "parts": [ SessionPartRecord( message_id=seed_user.message_id, session_id="sess-2", payload={ "text": "## Global Intent Summary\n做一个3d坦克大战游戏\n\n## Your Responsibility\nOwn the execution work item for role engineer." }, ) ], }, { "message": assistant, "parts": [SessionPartRecord(message_id=assistant.message_id, session_id="sess-2", payload={"text": "working on it"})], }, { "message": followup_user, "parts": [SessionPartRecord(message_id=followup_user.message_id, session_id="sess-2", payload={"text": "please keep the game browser-friendly"})], }, ] memory._get_visible_session_transcript = AsyncMock(return_value=visible_items) # type: ignore[method-assign] memory.build_session_memory_context = AsyncMock(return_value="") # type: ignore[method-assign] prompt_context = await memory.build_session_prompt_context( "sess-2", include_latest_user_turn=False, ) history_messages = await memory.build_session_history_messages( "sess-2", include_latest_user_turn=False, ) self.assertNotIn("Your Responsibility", prompt_context) self.assertNotIn("Global Intent Summary", prompt_context) self.assertEqual(len(history_messages), 1) self.assertIn("working on it", history_messages[0]["content"]) self.assertIn("working on it", prompt_context) # ---------------------------------------------------------------------- # SkillLibrary mode filtering and the playbook skill # ---------------------------------------------------------------------- class SkillModeFilteringTests(unittest.TestCase): def _make_library(self, *, with_playbook: bool = True) -> SkillLibrary: tmpdir = Path(tempfile.mkdtemp(prefix="opc-skill-test-")) skills_dir = tmpdir / "skills" skills_dir.mkdir() if with_playbook: playbook_dir = skills_dir / "collaboration-playbook" playbook_dir.mkdir() (playbook_dir / "SKILL.md").write_text( "---\n" "name: collaboration-playbook\n" "description: Standing rules for company-mode collaboration.\n" "always: true\n" "modes:\n" " - company_mode\n" "---\n\n" "# Collaboration Playbook\n\n" "Stay inside your work-item boundary.\n" ) memory_dir = skills_dir / "memory" memory_dir.mkdir() (memory_dir / "SKILL.md").write_text( "---\n" "name: memory\n" "description: Read and edit canonical OpenOPC memory.\n" "always: true\n" "---\n\n" "# Memory\n\n" "Use `.opc/memory/global.md` and `.opc/memory/projects/.md`.\n" ) # Also create one ordinary skill that should always show. ordinary_dir = skills_dir / "writing" ordinary_dir.mkdir() (ordinary_dir / "SKILL.md").write_text( "---\n" "name: writing\n" "description: Helpful writing patterns.\n" "---\n\n" "# Writing\n" ) library = SkillLibrary(tmpdir) library.load_all() return library def test_company_mode_skill_is_visible_under_company_mode(self) -> None: library = self._make_library() summary = library.build_skills_summary(execution_mode="company_mode") self.assertIn("## Skill: collaboration-playbook", summary) self.assertIn("Stay inside your work-item boundary.", summary) # Ordinary skill is also catalogued. self.assertIn("**writing**", summary) def test_company_mode_skill_is_completely_hidden_in_task_mode(self) -> None: library = self._make_library() summary = library.build_skills_summary(execution_mode="task_mode") # Body must be gone. self.assertNotIn("Stay inside your work-item boundary.", summary) self.assertNotIn("## Skill: collaboration-playbook", summary) # Description must ALSO be gone — even the catalog summary # entry must not appear in task mode. self.assertNotIn("collaboration-playbook", summary) self.assertNotIn( "Standing rules for company-mode collaboration.", summary ) # The ordinary skill is unaffected. self.assertIn("**writing**", summary) self.assertIn("## Skill: memory", summary) def test_company_mode_skill_is_hidden_when_execution_mode_unknown(self) -> None: library = self._make_library() summary = library.build_skills_summary(execution_mode=None) self.assertNotIn("collaboration-playbook", summary) self.assertIn("**writing**", summary) self.assertNotIn("## Skill: memory", summary) def test_memory_skill_visible_in_task_mode_for_all_agents(self) -> None: library = self._make_library() summary = library.build_skills_summary( execution_mode="task_mode", role_id="worker", user_facing=False, final_decider_role_id=None, ) self.assertIn("## Skill: memory", summary) def test_memory_skill_visible_only_to_user_facing_final_decider_in_company_mode(self) -> None: library = self._make_library() worker_summary = library.build_skills_summary( execution_mode="company_mode", role_id="worker", user_facing=True, final_decider_role_id="ceo", ) agent_turn_summary = library.build_skills_summary( execution_mode="company_mode", role_id="ceo", user_facing=False, final_decider_role_id="ceo", ) final_summary = library.build_skills_summary( execution_mode="company_mode", role_id="ceo", user_facing=True, final_decider_role_id="ceo", ) self.assertNotIn("## Skill: memory", worker_summary) self.assertNotIn("## Skill: memory", agent_turn_summary) self.assertIn("## Skill: memory", final_summary) def test_skill_visibility_helper_handles_modes_field_shapes(self) -> None: # Backward compat: skill with no modes is visible everywhere. skill_no_modes = Skill(name="x", modes=[]) self.assertTrue(SkillLibrary._skill_visible_in_mode(skill_no_modes, None)) self.assertTrue(SkillLibrary._skill_visible_in_mode(skill_no_modes, "company_mode")) self.assertTrue(SkillLibrary._skill_visible_in_mode(skill_no_modes, "task_mode")) # Restricted skill is hidden everywhere except declared modes. skill_company = Skill(name="y", modes=["company_mode"]) self.assertFalse(SkillLibrary._skill_visible_in_mode(skill_company, None)) self.assertFalse(SkillLibrary._skill_visible_in_mode(skill_company, "")) self.assertFalse(SkillLibrary._skill_visible_in_mode(skill_company, "task_mode")) self.assertTrue(SkillLibrary._skill_visible_in_mode(skill_company, "company_mode")) skill_memory = Skill(name="memory", modes=[]) self.assertTrue(SkillLibrary._skill_visible_in_mode(skill_memory, "task_mode")) self.assertFalse(SkillLibrary._skill_visible_in_mode(skill_memory, "company_mode")) self.assertTrue( SkillLibrary._skill_visible_in_mode( skill_memory, "company_mode", role_id="ceo", user_facing=True, final_decider_role_id="ceo", ) ) # ---------------------------------------------------------------------- # Real bundled collaboration-playbook skill # ---------------------------------------------------------------------- class BundledPlaybookSkillTests(unittest.TestCase): """Sanity-check the playbook shipped under .opc/skills/.""" @classmethod def setUpClass(cls) -> None: # tests/ -> repo root cls.repo_root = Path(__file__).resolve().parents[1] cls.skill_path = cls.repo_root / ".opc" / "skills" / "collaboration-playbook" / "SKILL.md" def test_playbook_file_exists_with_company_mode_only_frontmatter(self) -> None: self.assertTrue( self.skill_path.exists(), msg=f"Expected collaboration playbook at {self.skill_path}", ) text = self.skill_path.read_text(encoding="utf-8") self.assertIn("name: collaboration-playbook", text) self.assertIn("always: true", text) # The mode restriction is what makes the skill default-true # in company mode and invisible in task mode. self.assertIn("modes:", text) self.assertIn("- company_mode", text) def test_bundled_playbook_loads_through_skill_library(self) -> None: library = SkillLibrary(self.repo_root / ".opc") library.load_all() skill = library.get("collaboration-playbook") self.assertIsNotNone(skill) assert skill is not None self.assertTrue(skill.always) self.assertEqual(skill.modes, ["company_mode"]) self.assertIn("Stay inside your work-item boundary.", skill.content) company_summary = library.build_skills_summary(execution_mode="company_mode") self.assertIn("## Skill: collaboration-playbook", company_summary) task_summary = library.build_skills_summary(execution_mode="task_mode") self.assertNotIn("collaboration-playbook", task_summary) if __name__ == "__main__": unittest.main()