From b4d28aefeb1acb227765d6c1bad98a3c72bf44ce Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Tue, 7 Jul 2026 22:33:39 +0800 Subject: [PATCH] fix: stop misreading per-work-item spec as run-level company plan Every completed company session's follow-up was answered with the canned "Legacy company runtime run ... read-only" text (project 000, 2026-07-07). Root cause: snapshot loaders read task.metadata work_item_runtime_plan as a serialized run-level CompanyWorkItemRuntimePlan, but work-item tasks persist a per-item assignment spec (projection_id/turn_type/summary/deliverables/...) under that key. from_dict on the wrong shape silently yields an empty plan (no projections, empty metadata), _runtime_uses_multi_team_org returns False, and the resume path falls through to the legacy read-only branch. Add is_serialized_company_work_item_runtime_plan (a run-level plan always serializes with projections + runtime_model; a spec always carries projection_id) and route all full-plan metadata reads through serialized_company_plan_from_metadata, which skips wrong-shaped candidates so the loaders fall back to the sample-metadata-constructed plan instead of an empty one. Fixes existing DBs read-side; no data migration. Verified: old path on the 000 shape classifies multi_team_org=False, new path True; regression tests cover shape discrimination, snapshot classification, and the follow-up never reporting legacy. Co-Authored-By: Claude Fable 5 --- opc/engine.py | 11 +- opc/layer2_organization/company_mode.py | 40 ++++- tests/test_company_plan_metadata_shape.py | 203 ++++++++++++++++++++++ 3 files changed, 241 insertions(+), 13 deletions(-) create mode 100644 tests/test_company_plan_metadata_shape.py diff --git a/opc/engine.py b/opc/engine.py index 540a1fd..88de6ae 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -88,6 +88,7 @@ from opc.layer2_organization.company_mode import ( deserialize_company_work_item_runtime_plan, serialize_company_runtime_spec, serialize_company_work_item_runtime_plan, + serialized_company_plan_from_metadata, ) from opc.layer2_organization.company_runtime import canonical_role_session_id from opc.layer2_organization.metadata_ownership import ( @@ -4988,7 +4989,7 @@ class OPCEngine: plan_data = None for task in sorted(latest_by_projection_id.values(), key=lambda item: (item.created_at, item.id), reverse=True): - candidate = task.metadata.get("company_work_item_plan") or task.metadata.get("work_item_runtime_plan") + candidate = serialized_company_plan_from_metadata(task.metadata) if candidate: plan_data = candidate break @@ -6458,7 +6459,7 @@ class OPCEngine: for parent_session_id, group in runtime_groups.items(): plan_data = None for task in sorted(group, key=lambda item: (item.created_at, item.id), reverse=True): - candidate = task.metadata.get("company_work_item_plan") or task.metadata.get("work_item_runtime_plan") + candidate = serialized_company_plan_from_metadata(task.metadata) if candidate: plan_data = candidate break @@ -8112,11 +8113,7 @@ class OPCEngine: or task_metadata.get("company_profile", "") or getattr(self.config.org, "company_profile", "corporate") ).strip() or "corporate" - plan_payload = ( - task_metadata.get("company_work_item_plan") - or task_metadata.get("work_item_runtime_plan") - or {} - ) + plan_payload = serialized_company_plan_from_metadata(task_metadata) or {} work_item_plan = ( deserialize_company_work_item_runtime_plan(plan_payload) if isinstance(plan_payload, dict) and plan_payload diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 68a3c34..48a000c 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -317,6 +317,35 @@ def deserialize_company_work_item_runtime_plan(data: dict[str, Any] | None) -> C return deserialize_company_work_item_plan(data) +_SERIALIZED_PLAN_MARKER_KEYS = ("projections", "seeds", "root_projection_id", "runtime_model") + + +def is_serialized_company_work_item_runtime_plan(data: Any) -> bool: + """Whether ``data`` is a run-level serialized CompanyWorkItemRuntimePlan. + + Work-item tasks overload plan-adjacent metadata keys: ``work_item_runtime_plan`` + holds a per-item assignment spec (``projection_id``/``turn_type``/``summary``/...), + not the run-level plan. Feeding that shape to ``from_dict`` silently yields an + empty plan, which downstream misreads as a legacy (non multi-team-org) run and + refuses to resume the session. + """ + if not isinstance(data, dict) or not data: + return False + if "projection_id" in data: + return False + return any(key in data for key in _SERIALIZED_PLAN_MARKER_KEYS) + + +def serialized_company_plan_from_metadata(metadata: dict[str, Any] | None) -> dict[str, Any] | None: + """Return the first metadata value that is a real serialized run-level plan.""" + source = dict(metadata or {}) + for key in ("company_work_item_plan", "work_item_runtime_plan"): + candidate = source.get(key) + if is_serialized_company_work_item_runtime_plan(candidate): + return candidate + return None + + def _coerce_company_work_item_runtime_plan(plan: Any) -> CompanyWorkItemRuntimePlan | None: """Accept projection-plan-like test doubles without consuming obsolete plan fields.""" if plan is None or isinstance(plan, CompanyWorkItemRuntimePlan): @@ -1475,7 +1504,7 @@ class CompanyWorkItemExecutor: return plan, tasks plan_data = None for task in sorted(latest_by_projection_id.values(), key=lambda item: (item.created_at, item.id), reverse=True): - candidate = task.metadata.get("company_work_item_plan") or task.metadata.get("work_item_runtime_plan") + candidate = serialized_company_plan_from_metadata(task.metadata) if candidate: plan_data = candidate break @@ -1601,8 +1630,8 @@ class CompanyWorkItemExecutor: def _plan_view_for_task(self, task: Task) -> CompanyWorkItemRuntimePlan | None: if self._active_plan is not None: return self._active_plan - plan_data = task.metadata.get("company_work_item_plan") or task.metadata.get("work_item_runtime_plan") - if isinstance(plan_data, dict) and plan_data: + plan_data = serialized_company_plan_from_metadata(task.metadata) + if plan_data: return deserialize_company_work_item_runtime_plan(plan_data) return None @@ -8983,9 +9012,8 @@ class CompanyWorkItemExecutor: "requires_user_feedback": True, } plan_data = ( - intake_meta.get("company_work_item_plan") - or intake_meta.get("work_item_runtime_plan") - or dict(intake_meta.get("delegation_playbook", {}) or {}).get("company_work_item_plan") + serialized_company_plan_from_metadata(intake_meta) + or serialized_company_plan_from_metadata(dict(intake_meta.get("delegation_playbook", {}) or {})) ) if isinstance(plan_data, dict) and plan_data: try: diff --git a/tests/test_company_plan_metadata_shape.py b/tests/test_company_plan_metadata_shape.py new file mode 100644 index 0000000..e9a9136 --- /dev/null +++ b/tests/test_company_plan_metadata_shape.py @@ -0,0 +1,203 @@ +"""Regression tests for the company_work_item_plan / work_item_runtime_plan schema collision. + +Work-item tasks store a per-item assignment spec under ``work_item_runtime_plan`` +(keys: projection_id/turn_type/summary/deliverables/acceptance_criteria) while the +snapshot loaders used to read the same key as a serialized run-level +CompanyWorkItemRuntimePlan. Deserializing the spec shape silently produced an empty +plan, so every completed company session was misclassified as a legacy read-only run +and follow-up messages got the canned "Legacy company runtime run" reply +(project 000 forensics, 2026-07-07). +""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from opc.core.models import Task, TaskStatus +from opc.database.store import OPCStore +from opc.engine import OPCEngine +from opc.layer2_organization.company_mode import ( + is_serialized_company_work_item_runtime_plan, + serialize_company_work_item_runtime_plan, + serialized_company_plan_from_metadata, +) +from opc.layer2_organization.org_work_item_planner import ( + CompanyWorkItemRuntimePlan, + WorkItemProjectionSpec, +) + + +# Shape actually persisted on work-item tasks (verified against project 000 data). +SPEC_SHAPED_PLAN = { + "projection_id": "ceo::delivery::1899e271", + "turn_type": "deliver", + "summary": "Deliver the final result to the user.", + "inputs": ["Use the global intent summary as the mission baseline."], + "deliverables": ["Final delivery message."], + "acceptance_criteria": ["User receives the deliverable."], +} + + +def _full_plan() -> CompanyWorkItemRuntimePlan: + return CompanyWorkItemRuntimePlan( + profile="corporate", + projections=[ + WorkItemProjectionSpec( + projection_id="corporate::intake::ceo", + turn_type="intake", + title="CEO Intake", + summary="Frame the mission.", + role_id="ceo", + ) + ], + metadata={ + "execution_model": "multi_team_org", + "runtime_model": "multi_team_org", + }, + ) + + +class SerializedPlanShapeTests(unittest.TestCase): + def test_full_plan_serialization_is_recognized(self) -> None: + serialized = serialize_company_work_item_runtime_plan(_full_plan()) + self.assertTrue(is_serialized_company_work_item_runtime_plan(serialized)) + + def test_per_work_item_spec_is_rejected(self) -> None: + self.assertFalse(is_serialized_company_work_item_runtime_plan(SPEC_SHAPED_PLAN)) + + def test_non_dict_and_empty_are_rejected(self) -> None: + for value in (None, {}, [], "plan", 3): + self.assertFalse(is_serialized_company_work_item_runtime_plan(value)) + + def test_metadata_accessor_skips_spec_shape_and_null(self) -> None: + metadata = { + "company_work_item_plan": None, + "work_item_runtime_plan": dict(SPEC_SHAPED_PLAN), + } + self.assertIsNone(serialized_company_plan_from_metadata(metadata)) + + def test_metadata_accessor_prefers_canonical_key(self) -> None: + serialized = serialize_company_work_item_runtime_plan(_full_plan()) + metadata = { + "company_work_item_plan": serialized, + "work_item_runtime_plan": dict(SPEC_SHAPED_PLAN), + } + self.assertEqual(serialized_company_plan_from_metadata(metadata), serialized) + + def test_metadata_accessor_accepts_full_plan_in_fallback_key(self) -> None: + serialized = serialize_company_work_item_runtime_plan(_full_plan()) + metadata = {"work_item_runtime_plan": serialized} + self.assertEqual(serialized_company_plan_from_metadata(metadata), serialized) + + +class SnapshotClassificationTests(unittest.IsolatedAsyncioTestCase): + """A completed run whose tasks only carry spec-shaped plans must still classify + as a multi-team-org work-item run (never as a legacy read-only run).""" + + PARENT_SESSION = "sess-parent" + + async def _store(self) -> OPCStore: + tmpdir = tempfile.TemporaryDirectory() + self.addCleanup(tmpdir.cleanup) + store = OPCStore(Path(tmpdir.name) / "tasks.db") + await store.initialize() + self.addAsyncCleanup(store.close) + return store + + def _engine(self, store: OPCStore) -> OPCEngine: + engine = OPCEngine() + engine.project_id = "proj1" + engine.store = store + return engine + + def _work_item_task( + self, + task_id: str, + projection_id: str, + *, + status: TaskStatus, + turn_type: str = "execute", + extra_metadata: dict | None = None, + ) -> Task: + metadata = { + "company_profile": "corporate", + "execution_model": "multi_team_org", + "runtime_model": "multi_team_org", + "work_item_runtime": True, + "work_item_projection_id": projection_id, + "work_item_turn_type": turn_type, + # The 000 shape: canonical key null, fallback key holds the spec. + "company_work_item_plan": None, + "work_item_runtime_plan": dict(SPEC_SHAPED_PLAN, projection_id=projection_id), + } + metadata.update(extra_metadata or {}) + return Task( + id=task_id, + title=projection_id, + session_id=f"{self.PARENT_SESSION}:{task_id}", + parent_session_id=self.PARENT_SESSION, + status=status, + project_id="proj1", + metadata=metadata, + ) + + async def _seed_completed_run(self, store: OPCStore) -> None: + tasks = [ + self._work_item_task( + "intake-task", "corporate::intake::ceo", status=TaskStatus.DONE, turn_type="intake" + ), + self._work_item_task( + "execute-task", "cto::execute::d361bf3c", status=TaskStatus.DONE + ), + # Orphaned review attempt superseded by v2 (stuck non-terminal, as in 000). + self._work_item_task( + "review-v1-task", "review::467f36ff::v1", status=TaskStatus.PENDING, turn_type="review" + ), + self._work_item_task( + "delivery-task", + "ceo::delivery::1899e271", + status=TaskStatus.DONE, + turn_type="deliver", + extra_metadata={"feedback_closed": True, "feedback_scope": "final"}, + ), + ] + for task in tasks: + await store.save_task(task) + + async def test_snapshot_plan_classifies_as_multi_team_org(self) -> None: + store = await self._store() + await self._seed_completed_run(store) + engine = self._engine(store) + + snapshot = await engine._load_company_runtime_snapshot(self.PARENT_SESSION) + assert snapshot is not None + plan, tasks = snapshot + + self.assertTrue(engine._runtime_uses_multi_team_org(plan)) + self.assertTrue(tasks) + self.assertTrue(all(engine._task_uses_multi_team_org(task) for task in tasks)) + + async def test_followup_after_completed_run_never_reports_legacy(self) -> None: + store = await self._store() + await self._seed_completed_run(store) + engine = self._engine(store) + + class DummyCompanyExecutor: + async def execute(self, plan, tasks): # noqa: ANN001 + return "follow-up handled" + + engine.company_executor = DummyCompanyExecutor() + + response = await engine._maybe_resume_existing_company_runtime( + "where is the delivered file?", + session_id=self.PARENT_SESSION, + ) + + if response is not None: + self.assertNotIn("Legacy company runtime run", response) + + +if __name__ == "__main__": + unittest.main()