refactor(content): move orchestration markers off quick_context to typed jsonb

This commit is contained in:
Renn F
2026-06-21 02:48:07 +02:00
parent 7e74112fcc
commit 0fba4eed01
21 changed files with 270 additions and 316 deletions
+45 -25
View File
@@ -1,28 +1,50 @@
"""External-PR review dedup — review once per (project, PR, head commit).
``external_review_task_exists`` drives re-review off the PR's head SHA: an
unchanged PR (same head) is skipped, new commits (a new head SHA) open a fresh
review, and legacy/unknown-SHA tasks are never re-reviewed (no spam).
``external_review_task_exists`` drives re-review off the PR's head SHA, stored as
the ``external_pr_head`` orchestration marker (migration 041); dismissal is the
``dismissed`` marker. An unchanged PR (same head) is skipped, new commits open a
fresh review, and legacy/unknown-SHA tasks are never re-reviewed.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
import pytest
from roboco.foundation.policy.content import markers
from roboco.services.task import TaskService
def _service(quick_contexts: list[str | None]) -> TaskService:
"""A TaskService whose review-task query returns these quick_context values."""
def _service(scalar_rows: list[object]) -> TaskService:
"""A TaskService whose next query returns these scalar rows."""
res = MagicMock()
res.scalars.return_value.all.return_value = quick_contexts
res.scalars.return_value.all.return_value = scalar_rows
session = MagicMock()
session.execute = AsyncMock(return_value=res)
session.flush = AsyncMock()
return TaskService(session)
def _markers(head: str | None = None, dismissed: bool = False) -> dict:
om: dict = {}
if head is not None:
om["external_pr_head"] = head
if dismissed:
om["dismissed"] = True
return om
def _bind(svc: TaskService, name: str, value: object) -> None:
object.__setattr__(svc, name, value)
# ---------------------------------------------------------------------------
# external_review_task_exists — scalars are orchestration_markers dicts
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_no_task_yet_ingests() -> None:
svc = _service([])
@@ -31,14 +53,14 @@ async def test_no_task_yet_ingests() -> None:
@pytest.mark.asyncio
async def test_same_head_sha_skips() -> None:
svc = _service(["external_pr_head=abc"])
svc = _service([_markers("abc")])
assert await svc.external_review_task_exists(uuid4(), 170, "abc") is True
@pytest.mark.asyncio
async def test_new_head_sha_rereviews() -> None:
# PR got new commits since the last review → open a fresh review.
svc = _service(["external_pr_head=abc"])
svc = _service([_markers("abc")])
assert await svc.external_review_task_exists(uuid4(), 170, "def") is False
@@ -52,25 +74,26 @@ async def test_legacy_markerless_task_not_rereviewed() -> None:
@pytest.mark.asyncio
async def test_unknown_head_sha_does_not_spam() -> None:
# Can't detect change (no SHA from GitHub) → treat as reviewed.
svc = _service(["external_pr_head=abc"])
svc = _service([_markers("abc")])
assert await svc.external_review_task_exists(uuid4(), 170, None) is True
@pytest.mark.asyncio
async def test_multiple_old_shas_still_rereviews_new() -> None:
svc = _service(["external_pr_head=abc", "external_pr_head=def"])
svc = _service([_markers("abc"), _markers("def")])
assert await svc.external_review_task_exists(uuid4(), 170, "ghi") is False
assert await svc.external_review_task_exists(uuid4(), 170, "def") is True
def _bind(svc: TaskService, name: str, value: object) -> None:
object.__setattr__(svc, name, value)
# ---------------------------------------------------------------------------
# list queues — post-query dismissed filter (scalars are task objects)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_awaiting_decision_excludes_dismissed() -> None:
pending = MagicMock(quick_context="external_pr_head=abc")
dismissed = MagicMock(quick_context="external_pr_head=def dismissed=1")
pending = SimpleNamespace(orchestration_markers=_markers("abc"))
dismissed = SimpleNamespace(orchestration_markers=_markers("def", dismissed=True))
svc = _service([pending, dismissed])
out = await svc.list_external_pr_reviews_awaiting_decision()
assert out == [pending]
@@ -78,10 +101,8 @@ async def test_list_awaiting_decision_excludes_dismissed() -> None:
@pytest.mark.asyncio
async def test_list_external_pr_reviews_excludes_dismissed() -> None:
# The panel queue surfaces in-flight reviews too (the status filter lives in
# SQL); here we pin the post-query behavior: dismissed reviews drop out.
reviewing = MagicMock(quick_context="external_pr_head=abc")
dismissed = MagicMock(quick_context="external_pr_head=def dismissed=1")
reviewing = SimpleNamespace(orchestration_markers=_markers("abc"))
dismissed = SimpleNamespace(orchestration_markers=_markers("def", dismissed=True))
svc = _service([reviewing, dismissed])
out = await svc.list_external_pr_reviews()
assert out == [reviewing]
@@ -89,15 +110,14 @@ async def test_list_external_pr_reviews_excludes_dismissed() -> None:
@pytest.mark.asyncio
async def test_dismiss_marks_and_is_idempotent() -> None:
task = MagicMock(source="external_pr", quick_context="external_pr_head=abc")
session = MagicMock()
session.flush = AsyncMock()
svc = TaskService(session)
task = SimpleNamespace(source="external_pr", orchestration_markers=_markers("abc"))
svc = _service([])
_bind(svc, "get", AsyncMock(return_value=task))
await svc.dismiss_external_pr_review(uuid4())
assert "dismissed=1" in task.quick_context.split()
assert markers.is_dismissed(task) is True
await svc.dismiss_external_pr_review(uuid4()) # idempotent
assert task.quick_context.split().count("dismissed=1") == 1
assert markers.is_dismissed(task) is True
assert task.orchestration_markers["dismissed"] is True
@pytest.mark.asyncio
@@ -111,7 +131,7 @@ async def test_dismiss_rejects_non_external_pr() -> None:
# ---------------------------------------------------------------------------
# active_task_owns_branch — the internal-PR "is this a lifecycle PR?" check (#3)
# active_task_owns_branch — the internal-PR "is this a lifecycle PR?" check
# ---------------------------------------------------------------------------
+8 -4
View File
@@ -135,7 +135,7 @@ async def test_ingest_creates_review_task(db_session: AsyncSession) -> None:
assert task.task_type == TaskType.CODE
assert task.confirmed_by_human is False
assert task.status == TaskStatus.PENDING
assert task.quick_context == "external_pr_head=abc123"
assert task.orchestration_markers == {"external_pr_head": "abc123"}
@pytest.mark.asyncio
@@ -187,7 +187,7 @@ async def test_ingest_new_head_rereviews(db_session: AsyncSession) -> None:
await db_session.flush()
assert rereview is not None
assert rereview.quick_context == "external_pr_head=def456"
assert rereview.orchestration_markers == {"external_pr_head": "def456"}
reviews = await svc.list_external_pr_reviews()
matching = [t for t in reviews if t.pr_number == EXTERNAL_PR]
assert len(matching) == REVIEWS_AFTER_REREVIEW
@@ -376,8 +376,12 @@ async def test_find_supersede_umbrella_no_prefix_false_match(
assert five is not None
assert fifty is not None
assert UUID(str(five.id)) != UUID(str(fifty.id))
assert "pr=5 review=" in (five.quick_context or "")
assert "pr=50 review=" in (fifty.quick_context or "")
assert "pr=5 review=" in (five.orchestration_markers or {}).get(
"external_pr_supersede", ""
)
assert "pr=50 review=" in (fifty.orchestration_markers or {}).get(
"external_pr_supersede", ""
)
# ---------------------------------------------------------------------------
+35 -16
View File
@@ -1,36 +1,53 @@
"""required_cells decomposition gate — marker parse + uncovered-cell coverage.
The Main PM must create a subtask for each cell the brief explicitly names
(recorded as a ``required_cells:`` marker on the parent's quick_context). The
gate is inert when no marker is present, so legacy decompositions never block.
(recorded as a ``required_cells`` orchestration marker on the parent). The gate
is inert when no marker is present, so legacy decompositions never block.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
import pytest
from roboco.services.task import TaskService, extract_required_cells
def _task(required_cells: list[str] | None = None) -> SimpleNamespace:
om = {"required_cells": required_cells} if required_cells is not None else None
return SimpleNamespace(orchestration_markers=om)
# ---------------------------------------------------------------------------
# extract_required_cells (marker parser)
# extract_required_cells (marker reader)
# ---------------------------------------------------------------------------
def test_extract_required_cells_absent_is_empty() -> None:
assert extract_required_cells(None) == []
assert extract_required_cells("original_developer: abc\ndoc_notes: y") == []
assert extract_required_cells(_task()) == []
assert (
extract_required_cells(
SimpleNamespace(orchestration_markers={"original_developer": "abc"})
)
== []
)
def test_extract_required_cells_parses_and_normalizes() -> None:
qc = "original_developer: abc\nrequired_cells: Backend, Frontend , UX/UI"
assert extract_required_cells(qc) == ["backend", "frontend", "ux_ui"]
def test_extract_required_cells_normalizes() -> None:
assert extract_required_cells(_task(["Backend", "Frontend ", "UX/UI"])) == [
"backend",
"frontend",
"ux_ui",
]
def test_extract_required_cells_dedups_in_order() -> None:
out = extract_required_cells("required_cells: backend, backend, frontend")
assert out == ["backend", "frontend"]
assert extract_required_cells(_task(["backend", "backend", "frontend"])) == [
"backend",
"frontend",
]
# ---------------------------------------------------------------------------
@@ -38,10 +55,12 @@ def test_extract_required_cells_dedups_in_order() -> None:
# ---------------------------------------------------------------------------
def _service(parent_qc: str | None, child_teams: list[str | None]) -> TaskService:
def _service(
required_cells: list[str] | None, child_teams: list[str | None]
) -> TaskService:
"""A TaskService whose get()/get_subtasks() return a parent + these children."""
svc = TaskService(MagicMock())
parent = MagicMock(quick_context=parent_qc)
parent = _task(required_cells)
children = [MagicMock(team=t) for t in child_teams]
object.__setattr__(svc, "get", AsyncMock(return_value=parent))
object.__setattr__(svc, "get_subtasks", AsyncMock(return_value=children))
@@ -50,25 +69,25 @@ def _service(parent_qc: str | None, child_teams: list[str | None]) -> TaskServic
@pytest.mark.asyncio
async def test_uncovered_inert_without_marker() -> None:
svc = _service("doc_notes: x", ["backend"])
svc = _service(None, ["backend"])
assert await svc.uncovered_required_cells(uuid4()) == []
@pytest.mark.asyncio
async def test_uncovered_flags_the_dropped_cell() -> None:
# Brief named backend+frontend+ux_ui; only backend+frontend got subtasks.
svc = _service("required_cells: backend, frontend, ux_ui", ["backend", "frontend"])
svc = _service(["backend", "frontend", "ux_ui"], ["backend", "frontend"])
assert await svc.uncovered_required_cells(uuid4()) == ["ux_ui"]
@pytest.mark.asyncio
async def test_uncovered_empty_when_all_named_cells_covered() -> None:
svc = _service("required_cells: backend, frontend", ["frontend", "backend"])
svc = _service(["backend", "frontend"], ["frontend", "backend"])
assert await svc.uncovered_required_cells(uuid4()) == []
@pytest.mark.asyncio
async def test_uncovered_normalizes_child_team_form() -> None:
# Marker uses underscore, child team uses the slash form — they match.
svc = _service("required_cells: ux_ui", ["UX/UI"])
svc = _service(["ux_ui"], ["UX/UI"])
assert await svc.uncovered_required_cells(uuid4()) == []
@@ -156,7 +156,7 @@ async def test_originate_creates_pending_main_pm_assigned_task(
assert task.team == Team.MAIN_PM
assert task.source == "self_heal"
assert task.acceptance_criteria # non-empty (AC-guardrail)
assert "self_heal_fp=" in (task.quick_context or "")
assert (task.orchestration_markers or {}).get("self_heal_fp")
@pytest.mark.asyncio
+34 -55
View File
@@ -1,21 +1,15 @@
"""Supersede umbrella close-on-land guards.
"""Supersede umbrella close-on-land guards (orchestration-marker storage).
Covers the parts of the external-PR supersede flow that decide whether and
which a landed supersede's contributor PR gets retired:
- ``supersede_marker_line`` anchors marker/state checks to the marker line, so
free-form CEO escalation/approval notes appended to the same multi-writer
``quick_context`` can't be mistaken for the marker.
- ``supersede_umbrellas_pending_close`` only returns umbrellas whose
replacement work actually landed (a non-cancelled descendant carrying a PR),
not every COMPLETED umbrella the CEO can force-complete over a cancelled
code subtask.
- ``mark_supersede_pr_closed`` writes the ``closed=1`` idempotency token onto
the marker line, surviving appended notes.
The supersede marker (``pr={n} review={uuid}`` plus a ``closed=1`` token once
the contributor PR is retired) lives in ``orchestration_markers`` after
migration 041 isolated from the human ``quick_context``, so CEO escalation /
approval notes can no longer be mistaken for it.
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
@@ -23,11 +17,10 @@ import pytest
from roboco.models.base import TaskStatus
from roboco.services.task import TaskService, supersede_marker_line
_MARKER = "external_pr_supersede pr=5 review=abc"
_VALUE = "pr=5 review=abc"
def _scalars_all(rows: list[object]) -> MagicMock:
"""A session.execute return value whose .scalars().all() yields `rows`."""
res = MagicMock()
res.scalars.return_value.all.return_value = rows
return res
@@ -44,25 +37,23 @@ def _bind(svc: TaskService, name: str, value: object) -> None:
object.__setattr__(svc, name, value)
def _task(supersede: str | None = None, **kw: Any) -> SimpleNamespace:
om = {"external_pr_supersede": supersede} if supersede is not None else None
return SimpleNamespace(orchestration_markers=om, **kw)
# ---------------------------------------------------------------------------
# supersede_marker_line — line anchoring
# supersede_marker_line — reads the marker value
# ---------------------------------------------------------------------------
def test_marker_line_returns_marker_ignoring_appended_notes() -> None:
qc = f"{_MARKER}\nceo_approval_notes: shipped, looks good"
assert supersede_marker_line(qc) == _MARKER
def test_marker_line_not_fooled_by_closed_token_in_note() -> None:
qc = f"{_MARKER}\nceo_approval_notes: marked closed=1 in jira"
# The marker line itself carries no closed=1, so the PR is NOT yet closed.
assert "closed=1" not in supersede_marker_line(qc).split()
def test_marker_line_returns_value() -> None:
assert supersede_marker_line(_task(_VALUE)) == _VALUE
def test_marker_line_empty_when_absent() -> None:
assert supersede_marker_line("no marker here\nescalation_notes: x") == ""
assert supersede_marker_line(None) == ""
assert supersede_marker_line(_task()) == ""
assert supersede_marker_line(SimpleNamespace(orchestration_markers=None)) == ""
# ---------------------------------------------------------------------------
@@ -72,29 +63,25 @@ def test_marker_line_empty_when_absent() -> None:
@pytest.mark.asyncio
async def test_pending_close_excludes_umbrella_with_closed_marker() -> None:
umbrella = MagicMock(id=uuid4(), quick_context=f"{_MARKER} closed=1")
umbrella = _task(f"{_VALUE} closed=1", id=uuid4())
svc = _service(_scalars_all([umbrella]))
_bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=True))
assert await svc.supersede_umbrellas_pending_close() == []
@pytest.mark.asyncio
async def test_pending_close_keeps_umbrella_with_closed_token_only_in_note() -> None:
# A CEO note containing the literal "closed=1" must NOT retire the PR.
umbrella = MagicMock(
id=uuid4(), quick_context=f"{_MARKER}\nceo_approval_notes: closed=1 elsewhere"
)
async def test_pending_close_keeps_open_landed_umbrella() -> None:
umbrella = _task(_VALUE, id=uuid4())
svc = _service(_scalars_all([umbrella]))
_bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=True))
out = await svc.supersede_umbrellas_pending_close()
assert out == [umbrella]
assert await svc.supersede_umbrellas_pending_close() == [umbrella]
@pytest.mark.asyncio
async def test_pending_close_requires_landed_replacement() -> None:
# COMPLETED + no closed marker, but the replacement never landed (the code
# subtask was cancelled) — close-on-land must skip it.
umbrella = MagicMock(id=uuid4(), quick_context=_MARKER)
umbrella = _task(_VALUE, id=uuid4())
svc = _service(_scalars_all([umbrella]))
_bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=False))
assert await svc.supersede_umbrellas_pending_close() == []
@@ -116,7 +103,6 @@ async def test_replacement_landed_true_for_completed_descendant_with_pr() -> Non
async def test_replacement_landed_false_when_descendant_cancelled() -> None:
child = MagicMock(id=uuid4(), status=TaskStatus.CANCELLED, pr_number=42)
svc = _service(_scalars_all([child]))
# The `seen` guard terminates the walk even though the mock re-returns child.
assert await svc._supersede_replacement_landed(uuid4()) is False
@@ -128,44 +114,37 @@ async def test_replacement_landed_false_when_completed_without_pr() -> None:
# ---------------------------------------------------------------------------
# find_supersede_umbrella — marker-line dedup
# find_supersede_umbrella — value dedup (pr=N review= prefix match)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_find_umbrella_matches_marker_not_note() -> None:
match = MagicMock(id=uuid4(), quick_context=f"{_MARKER}\nescalation_notes: x")
other = MagicMock(
id=uuid4(),
# marker for a different PR, but a note mentions "pr=5 review=" text
quick_context="external_pr_supersede pr=9 review=z\nnote: see pr=5 review= ok",
)
async def test_find_umbrella_matches_by_value() -> None:
match = _task(_VALUE, id=uuid4())
other = _task("pr=9 review=z", id=uuid4()) # different PR
svc = _service(_scalars_all([other, match]))
found = await svc.find_supersede_umbrella(uuid4(), 5)
assert found is match
# ---------------------------------------------------------------------------
# mark_supersede_pr_closed — token written on the marker line
# mark_supersede_pr_closed — token appended to the marker value
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_mark_closed_appends_token_to_marker_line() -> None:
task = MagicMock(quick_context=f"{_MARKER}\nceo_approval_notes: shipped")
async def test_mark_closed_appends_token() -> None:
task = _task(_VALUE)
svc = _service(_scalars_all([]))
_bind(svc, "get", AsyncMock(return_value=task))
await svc.mark_supersede_pr_closed(uuid4())
lines = task.quick_context.splitlines()
assert lines[0] == f"{_MARKER} closed=1"
assert lines[1] == "ceo_approval_notes: shipped" # note untouched
assert supersede_marker_line(task) == f"{_VALUE} closed=1"
@pytest.mark.asyncio
async def test_mark_closed_is_idempotent_on_marker_line() -> None:
task = MagicMock(quick_context=f"{_MARKER} closed=1\nceo_approval_notes: x")
async def test_mark_closed_is_idempotent() -> None:
task = _task(f"{_VALUE} closed=1")
svc = _service(_scalars_all([]))
_bind(svc, "get", AsyncMock(return_value=task))
await svc.mark_supersede_pr_closed(uuid4())
# No second closed=1 token appended.
assert task.quick_context.splitlines()[0] == f"{_MARKER} closed=1"
assert supersede_marker_line(task) == f"{_VALUE} closed=1"