From eb0dcb6ecbf141a2160c5bd8a50f7707544f2be4 Mon Sep 17 00:00:00 2001 From: Renzo F <45401804+rennf93@users.noreply.github.com> Date: Fri, 24 Jul 2026 17:20:20 +0200 Subject: [PATCH] fix(orchestrator): task-scoped oscillation breaker for escalate/unblock ping-pong (#685) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(orchestrator): task-scoped oscillation breaker for escalate/unblock ping-pong An escalation ping-pong oscillates a task between two agents (cell PM escalate_up -> BLOCKED -> main PM unblock -> restored -> respawn -> escalate again). The per-(agent, task) respawn gate never trips on it: the restored side is dispatched by _dispatch_claimed_without_agent, which consults no respawn counter at all, so one side of the round trip always has fuel regardless of the other's strikes — and even a tripped main-PM counter only stalls the task silently at blocked instead of surfacing the oscillation. - Strikes are counted task-scoped at the unblock() chokepoint (agent-agnostic; legitimate needs_revision rework never calls unblock, so it structurally cannot trip this), durable in the existing orchestration_markers column — no migration. - Progress between round-trips (commits / revision_count advancing) resets the count: real forward motion is not an oscillation. - On trip: the task is blocked with a HUMAN resolver (the budget-breach posture), both dispatchers stop respawning onto it, further unblock() refuses until an admin override clears the marker, and the CEO notification names both agents and the cycle count. - _notification_has_live_work now treats a HITL-blocked related task as no live work, closing the same loop for the admin-route escalation path. * fix(orchestrator): wire the oscillation trip to the dispatchers and make recovery reachable - TaskResponse serializes blocker_resolver_type: the dispatchers' HITL-blocked skip and the notification-path live-work check now actually fire over the wire instead of only against in-process rows. - The oscillation marker clears on every human transition out of BLOCKED (snapshot or not), and the human unblock route treats a tripped task as the requested intervention: clears the marker and proceeds, while the agent gateway verb keeps refusing. - The progress fingerprint includes the terminal-children count, so a coordination root whose children advanced between escalations resets instead of accruing toward a false trip. --------- Co-authored-by: Renn F --- roboco/api/schemas/tasks.py | 5 + roboco/foundation/policy/content/markers.py | 74 ++++++ roboco/runtime/orchestrator.py | 32 ++- .../services/gateway/choreographer/_impl.py | 241 ++++++++++++++---- roboco/services/notification.py | 53 ++++ roboco/services/task.py | 47 +++- tests/unit/api/test_schemas_tasks.py | 96 ++++++- .../foundation/policy/content/test_markers.py | 45 ++++ .../unit/gateway/test_oscillation_breaker.py | 222 ++++++++++++++++ .../test_blocker_and_claimed_dispatch.py | 94 +++++++ tests/unit/services/test_task.py | 147 +++++++++++ 11 files changed, 990 insertions(+), 66 deletions(-) create mode 100644 tests/unit/gateway/test_oscillation_breaker.py diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index 3aec8a11..d273267c 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -309,6 +309,10 @@ class TaskResponse(BaseModel): # Status status: TaskStatus + # Who resolves a `blocked` task: AGENT (respawn as normal) or HUMAN (the + # dispatchers must skip it — see orchestrator._is_hitl_blocked). None + # outside `blocked`. + blocker_resolver_type: BlockerResolverType | None = None priority: int sequence: int # Order number within siblings # Cost cap (ROBOCO_TASK_BUDGETS_ENABLED). Null = no cap (explicit-input only). @@ -891,6 +895,7 @@ def task_to_response(task: "TaskTable") -> TaskResponse: constraints=getattr(task, "constraints", None), acceptance_criteria=task.acceptance_criteria or [], status=task.status, + blocker_resolver_type=task.blocker_resolver_type, priority=task.priority, sequence=task.sequence, budget_usd=getattr(task, "budget_usd", None), diff --git a/roboco/foundation/policy/content/markers.py b/roboco/foundation/policy/content/markers.py index 320f27b0..c0ed2998 100644 --- a/roboco/foundation/policy/content/markers.py +++ b/roboco/foundation/policy/content/markers.py @@ -542,3 +542,77 @@ def is_budget_blocked(task: HasMarkers) -> bool: def clear_budget_blocked(task: HasMarkers) -> None: clear_marker(task, BUDGET_BLOCKED) + + +# --- escalate_up/unblock oscillation breaker ------------------------------- +# A round trip (escalate_up blocks the task, unblock restores it) that keeps +# repeating with nothing landing in between is a deadlock, not rework — and +# the orchestrator's per-(agent, task) respawn breaker misses it structurally: +# the escalator and the resolver each own only half the cycle's spawns, so +# neither agent's own counter accrues at the cycle's real rate. `unblock` +# stamps this marker on every restore with a cheap progress fingerprint +# (commit count + revision_count, both already loaded on the task row, plus +# one COUNT query for terminal children — a PM coordination root never +# commits itself, so child completions are its only progress signal); an +# unchanged fingerprint accrues a strike, any change (real forward motion +# between escalations) resets to 1. Past the trip threshold the task is +# force-BLOCKED for a human instead of restored, and `unblock` refuses every +# further call on it until a human clears it — either an admin status +# override out of BLOCKED (`TaskService._admin_out_of_blocked`, snapshot or +# not) or the legacy human/panel unblock route (`TaskService.unblock`). +# Payload: {"strikes": int, "progress_fp": [int, int, int], "tripped": bool}. + +OSCILLATION_STRIKES = "oscillation_strikes" + + +def get_oscillation_strikes(task: HasMarkers) -> int: + val = get_marker(task, OSCILLATION_STRIKES) + strikes = val.get("strikes") if isinstance(val, dict) else None + return int(strikes) if isinstance(strikes, int) else 0 + + +def _get_oscillation_progress_fp(task: HasMarkers) -> list[int] | None: + val = get_marker(task, OSCILLATION_STRIKES) + fp = val.get("progress_fp") if isinstance(val, dict) else None + return list(fp) if isinstance(fp, list) else None + + +def is_oscillation_tripped(task: HasMarkers) -> bool: + val = get_marker(task, OSCILLATION_STRIKES) + return bool(val.get("tripped")) if isinstance(val, dict) else False + + +def bump_oscillation_strikes(task: HasMarkers, progress_fp: list[int]) -> int: + """Record one restore cycle against ``progress_fp``; returns the new + strike count. A fingerprint that differs from the last recorded one + resets to 1 (real progress happened between escalations); the first-ever + call (no prior fingerprint) and a repeated, unchanged fingerprint both + accrue from wherever the counter already was.""" + prior_fp = _get_oscillation_progress_fp(task) + strikes = ( + get_oscillation_strikes(task) + 1 + if prior_fp is None or prior_fp == progress_fp + else 1 + ) + set_marker( + task, + OSCILLATION_STRIKES, + { + "strikes": strikes, + "progress_fp": progress_fp, + "tripped": is_oscillation_tripped(task), + }, + ) + return strikes + + +def mark_oscillation_tripped(task: HasMarkers) -> None: + set_marker( + task, + OSCILLATION_STRIKES, + { + "strikes": get_oscillation_strikes(task), + "progress_fp": _get_oscillation_progress_fp(task) or [], + "tripped": True, + }, + ) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 7e45987c..20fd8c98 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -4831,29 +4831,32 @@ class AgentOrchestrator: return None return dt if dt.tzinfo else dt.replace(tzinfo=UTC) - async def _fetch_task_status( + async def _fetch_task_fields( self, client: httpx.AsyncClient, task_id: str - ) -> str | None: - """Best-effort GET /tasks/{id} → status string (None on any failure — + ) -> dict[str, Any] | None: + """Best-effort GET /tasks/{id} → full task dict (None on any failure — fail-open so a fetch hiccup never suppresses a real escalation).""" try: resp = await client.get(f"{self._api_url}/tasks/{task_id}") if resp.status_code == http_status.HTTP_200_OK: - status = resp.json().get("status") - return str(status) if status is not None else None + return cast("dict[str, Any]", resp.json()) except Exception as exc: - logger.debug("notification task-status fetch failed", error=str(exc)) + logger.debug("notification task-fields fetch failed", error=str(exc)) return None async def _notification_has_live_work( self, client: httpx.AsyncClient, notif: dict[str, Any] ) -> bool: """False when a notification has no live work behind it — the 'is there - actually something to do' gate for notification-triggered spawns. Three + actually something to do' gate for notification-triggered spawns. Four obvious markers: it has expired, it is stale past the spawn-age window - (wedged / reloaded from before a restart), or its related task is - already terminal (the work is done). Fail-open: an unparseable field or - a failed task fetch never suppresses a spawn. + (wedged / reloaded from before a restart), its related task is already + terminal (the work is done), or that task is HITL-blocked (a human, + not a respawn, resolves it — e.g. the oscillation breaker tripped and + force-blocked it; the task-status dispatchers already skip these via + ``_is_hitl_blocked``, but this notification-driven path carries no + task-status gate of its own). Fail-open: an unparseable field or a + failed task fetch never suppresses a spawn. """ now = datetime.now(UTC) expires = self._parse_iso_dt(notif.get("expires_at")) @@ -4865,9 +4868,12 @@ class AgentOrchestrator: return False task_id = notif.get("related_task_id") if task_id: - status = await self._fetch_task_status(client, str(task_id)) - if status in ("completed", "cancelled"): - return False + fields = await self._fetch_task_fields(client, str(task_id)) + if fields is not None: + if fields.get("status") in ("completed", "cancelled"): + return False + if self._is_hitl_blocked(fields): + return False return True def _is_parallel_phase_claim( diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index db0ab373..184913ba 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -110,6 +110,16 @@ _PM_SUBTASKS_MAX = 7 # forward progress before anyone noticed). _BLOCK_FLIP_NOTIFY_THRESHOLD = 3 +# unblock's oscillation breaker: past the (higher) notify threshold above, a +# cycle that keeps repeating with NO progress between rounds (see +# markers.bump_oscillation_strikes) is force-blocked for a human instead of +# restored again — the notify threshold alone only alerts, it never stops the +# respawns (live incident: an escalate_up/unblock cycle split across two +# agents' per-(agent, task) respawn counters, neither of which accrued at the +# cycle's real rate, burned spawns for hours with the CEO alert already fired +# and ignored/unactioned). +_OSCILLATION_TRIP_THRESHOLD = 5 + def _thin_subtask_hint(sub_tasks: list[Any]) -> str | None: """Return a hint if any PM sub_task is title-only / thin / over-long. @@ -7022,6 +7032,61 @@ class Choreographer: context_briefing=await self._briefing_for(pm_agent_id, None), ) + async def _unblock_preflight_guards( + self, pm_agent_id: UUID, task_id: UUID, t: Any, role: str + ) -> Envelope | None: + """Verb-specific preflight gates for ``unblock``, checked in order; + the first rejection wins, ``None`` means proceed. Factored out to + keep ``unblock`` under the return-count budget as gates accrete + (status / dependency / budget / oscillation). + """ + if str(t.status) != "blocked": + return Envelope.invalid_state( + message=f"task {task_id} is in {t.status}, expected blocked", + remediate=( + "this task is not blocked; call triage() to find blocked tasks" + ), + context_briefing=await self._briefing_for(pm_agent_id, task_id), + ).with_introspection(task=t, role=role) + + # A dependency block must not be cleared by hand. It auto-clears via + # _unblock_dependents the moment its last dependency reaches a terminal + # state; forcing it now would let the dependent proceed without the + # upstream's work (e.g. a frontend task built before its UX design lands). + dep_ids = list(t.dependency_ids or []) + unmet = await self.task.unmet_dependency_ids(dep_ids) if dep_ids else [] + if unmet: + return Envelope.invalid_state( + message=( + f"task {task_id} still depends on {len(unmet)} " + "unfinished task(s); a dependency block clears on its " + "own once the upstream work completes" + ), + remediate=( + "don't force this — let the dependency finish; the task " + "auto-unblocks the moment its last dependency reaches " + "completed/cancelled" + ), + context_briefing=await self._briefing_for(pm_agent_id, task_id), + ).with_introspection(task=t, role=role) + + # Budget-breach block: re-check spend-vs-cap before letting the PM + # clear it. Without this, a PM unblock on a task the orchestrator's + # sweep blocked for a $ overrun would silently re-breach the same cap + # the very next tick if the CEO's raise didn't actually clear it (or + # never raised it at all). + if guard := await self._budget_unblock_guard(t): + return guard.with_introspection(task=t, role=role) + + # Oscillation breaker: once tripped (an escalate_up/unblock round + # trip repeated past the threshold with no progress between rounds), + # only a human admin override clears it — see + # _oscillation_unblock_guard / _maybe_trip_oscillation_breaker. + if guard := self._oscillation_unblock_guard(t): + return guard.with_introspection(task=t, role=role) + + return None + async def unblock( self, pm_agent_id: UUID, task_id: UUID, reason: str, *, restore: bool = True ) -> Envelope: @@ -7038,57 +7103,9 @@ class Choreographer: ) agent = await self.task.agent_for(pm_agent_id) role = str(agent.role) if agent is not None else "cell_pm" - if str(t.status) != "blocked": + if env := await self._unblock_preflight_guards(pm_agent_id, task_id, t, role): return await self._emit_rejection( - Envelope.invalid_state( - message=f"task {task_id} is in {t.status}, expected blocked", - remediate=( - "this task is not blocked; call triage() to find blocked tasks" - ), - context_briefing=await self._briefing_for(pm_agent_id, task_id), - ).with_introspection(task=t, role=role), - agent_id=pm_agent_id, - task_id=task_id, - verb="unblock", - ) - - # A dependency block must not be cleared by hand. It auto-clears via - # _unblock_dependents the moment its last dependency reaches a terminal - # state; forcing it now would let the dependent proceed without the - # upstream's work (e.g. a frontend task built before its UX design lands). - dep_ids = list(t.dependency_ids or []) - unmet = await self.task.unmet_dependency_ids(dep_ids) if dep_ids else [] - if unmet: - return await self._emit_rejection( - Envelope.invalid_state( - message=( - f"task {task_id} still depends on {len(unmet)} " - "unfinished task(s); a dependency block clears on its " - "own once the upstream work completes" - ), - remediate=( - "don't force this — let the dependency finish; the task " - "auto-unblocks the moment its last dependency reaches " - "completed/cancelled" - ), - context_briefing=await self._briefing_for(pm_agent_id, task_id), - ).with_introspection(task=t, role=role), - agent_id=pm_agent_id, - task_id=task_id, - verb="unblock", - ) - - # Budget-breach block: re-check spend-vs-cap before letting the PM - # clear it. Without this, a PM unblock on a task the orchestrator's - # sweep blocked for a $ overrun would silently re-breach the same cap - # the very next tick if the CEO's raise didn't actually clear it (or - # never raised it at all). - if guard := await self._budget_unblock_guard(t): - return await self._emit_rejection( - guard.with_introspection(task=t, role=role), - agent_id=pm_agent_id, - task_id=task_id, - verb="unblock", + env, agent_id=pm_agent_id, task_id=task_id, verb="unblock" ) # Write-then-gate: the PM's unblock reason is recorded as the @@ -7106,8 +7123,15 @@ class Choreographer: verb="unblock", ) + # Captured before the restore below clears it (_restore_block_ownership) + # — the oscillation-trip notification names both sides of the round trip. + escalator_id = t.blocker_raised_by t = await self.task.unblock_with_restore(pm_agent_id, task_id, restore=restore) await self._maybe_notify_block_flip(task_id, t, t.title) + if tripped := await self._maybe_trip_oscillation_breaker( + task_id, t, escalator_id=escalator_id, resolver_id=pm_agent_id + ): + return tripped.with_introspection(task=t, role=role) next_msg = ( "task restored to its pre-block state — original assignee will resume" if restore @@ -7157,6 +7181,123 @@ class Choreographer: flip_count=flip_count, ) + def _oscillation_unblock_guard(self, t: Any) -> Envelope | None: + """Refuse ``unblock`` once the oscillation breaker has tripped. + + Unlike the budget guard (``_budget_unblock_guard``) there is no live + condition to re-check here (no "spend dropped below cap" analogue) — + a human must actively resolve this: reassign the task, fix whatever + the escalation kept citing, or cancel it. An admin status override + (``admin_set_status`` out of BLOCKED) bypasses this guard entirely and + clears the marker as part of that restore, so the human path stays + open regardless. + """ + if not markers.is_oscillation_tripped(t): + return None + return Envelope.invalid_state( + message=( + f"task {t.id} was force-blocked: its escalate_up/unblock " + "cycle repeated with no progress between rounds" + ), + remediate=( + "this needs a human to resolve — reassign the task, fix the " + "underlying blocker, or cancel it; unblock() will keep " + "refusing until an admin moves the task out of blocked" + ), + ) + + async def _maybe_trip_oscillation_breaker( + self, + task_id: UUID, + t: Any, + *, + escalator_id: Any, + resolver_id: UUID, + ) -> Envelope | None: + """Force-BLOCK a task whose escalate/unblock cycle keeps repeating + with no progress; returns the envelope to hand back on a fresh trip, + else ``None``. + + The progress fingerprint is deliberately cheap — commit count + + revision_count are already loaded on ``t``, plus one COUNT query for + terminal children. The child count matters because a PM coordination + root never commits itself (all real progress lands on child rows), so + the first two components alone are structurally static on one — a + busy root would false-trip on its 6th lifetime legitimate escalation + even with children completing between rounds. Findings only move via + QA/PR-review/PM-reject, a different lifecycle path from + escalate_up/unblock entirely, so they carry no signal here. + """ + terminal_children = await self.task.terminal_children_count(task_id) + progress_fp = [ + len(t.commits or []), + int(t.revision_count or 0), + terminal_children, + ] + strikes = markers.bump_oscillation_strikes(t, progress_fp) + if strikes <= _OSCILLATION_TRIP_THRESHOLD: + return None + from roboco.models.base import BlockerResolverType, TaskStatus + + try: + t.blocker_resolver_type = BlockerResolverType.HUMAN + await self.task.admin_set_status( + task_id, TaskStatus.BLOCKED, actor_role="system" + ) + markers.mark_oscillation_tripped(t) + except Exception: + logger.warning( + "failed to force-block oscillating task", + task_id=str(task_id), + strikes=strikes, + ) + return None + await self._notify_ceo_oscillation( + task_id, + strikes, + t.title, + escalator_id=escalator_id, + resolver_id=resolver_id, + ) + return Envelope.ok( + status=str(t.status), + task_id=str(task_id), + next=( + "this task's escalate/unblock cycle repeated with no progress " + "and was force-blocked for a human to resolve — do not call " + "unblock again; wait for the CEO" + ), + ) + + async def _notify_ceo_oscillation( + self, + task_id: UUID, + strikes: int, + task_title: str | None, + *, + escalator_id: Any, + resolver_id: UUID, + ) -> None: + """Best-effort CEO alert when the oscillation breaker trips; never + raises — the block itself already happened regardless.""" + from roboco.services.notification import NotificationService + + try: + await NotificationService().send_oscillation_blocked_notification( + task_id=str(task_id), + strikes=strikes, + escalator=escalator_id, + resolver=resolver_id, + db_session=self.task.session, + task_title=task_title, + ) + except Exception: + logger.warning( + "failed to send CEO oscillation-blocked notification", + task_id=str(task_id), + strikes=strikes, + ) + async def _own_review_hint(self, pm_agent_id: UUID, exclude_task_id: UUID) -> str: """Remediate suffix naming the PM's OWN task ready to complete. diff --git a/roboco/services/notification.py b/roboco/services/notification.py index 5e983012..2d786e8f 100644 --- a/roboco/services/notification.py +++ b/roboco/services/notification.py @@ -168,6 +168,59 @@ class NotificationService: ) ) + async def send_oscillation_blocked_notification( + self, + task_id: str, + strikes: int, + escalator: str | UUID | None, + resolver: str | UUID | None, + to_agent: str = "ceo", + db_session: AsyncSession | None = None, + task_title: str | None = None, + ) -> None: + """Alert the CEO that an escalate/unblock cycle was force-blocked. + + Raised from the choreographer's ``unblock`` once the round-trip + counter (progress-discriminated — see + ``markers.bump_oscillation_strikes``) crosses the trip threshold: the + task keeps bouncing between the same two agents (one escalates, the + other unblocks) with nothing landing in between, so automatic + recovery hands it to a human instead of burning more spawns. + """ + logger.info( + "Sending oscillation-blocked notification", + task_id=task_id, + strikes=strikes, + escalator=str(escalator) if escalator else None, + resolver=str(resolver) if resolver else None, + ) + display = task_display(task_title, task_id) + escalator_label = await agent_display(escalator, db_session) + resolver_label = await agent_display(resolver, db_session) + who = ( + " and ".join(s for s in (escalator_label, resolver_label) if s) + or "its agents" + ) + body = ( + f"Task {display} bounced between {who} {strikes} times " + "(escalate_up / unblock) with no progress in between and has " + "been force-blocked for a human to resolve. Automatic respawns " + "are stopped for this task — reassign it, fix the root cause, " + "or cancel it." + ) + await self._create_notification( + CreateNotificationParams( + notification_type=NotificationType.BLOCKER_ESCALATION, + priority=NotificationPriority.HIGH, + from_agent="system", + to_agents=[to_agent], + subject=f"Task {display} oscillation-blocked ({strikes}x)", + body=body, + related_task_id=task_id, + ), + db_session=db_session, + ) + async def send_qa_ready_notification( self, task_id: str, diff --git a/roboco/services/task.py b/roboco/services/task.py index 95705a02..891205d2 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -5248,6 +5248,19 @@ class TaskService(BaseService): ) return task + def _clear_tripped_oscillation_marker(self, task: TaskTable) -> None: + """The legacy human/panel unblock route never goes through the + gateway's `_oscillation_unblock_guard` — reaching a tripped task + here IS the human intervention the breaker demands, so clear it + rather than leave it to haunt the task's next legitimate + block/unblock cycle. The agent-verb `unblock` gateway path never + reaches here with the marker still tripped: its own guard already + refused before this method runs. A no-op when not tripped, so an + in-flight (non-tripped) strike count survives an ordinary unblock. + """ + if markers.is_oscillation_tripped(task): + markers.clear_marker(task, markers.OSCILLATION_STRIKES) + async def unblock( self, task_id: UUID, agent_role: str | None = None ) -> TaskTable | None: @@ -5289,6 +5302,7 @@ class TaskService(BaseService): task.claimed_by = owner # Clear resolver metadata — only meaningful while BLOCKED. task.blocker_resolver_type = None + self._clear_tripped_oscillation_marker(task) # A task with a branch was claimed before it blocked, so resume it # in_progress. A task with NO branch was blocked before it was ever # claimed (e.g. a dependency-gated claim that got escalated); it cannot @@ -9694,6 +9708,22 @@ class TaskService(BaseService): statuses = result.scalars().all() return all(s in terminal for s in statuses) + async def terminal_children_count(self, task_id: UUID) -> int: + """Count of direct subtasks in a terminal status (COMPLETED/CANCELLED). + + One cheap COUNT query — feeds the oscillation breaker's progress + fingerprint, since a PM coordination root never commits itself (all + real progress lands on child rows); ``unblock`` is rare enough that + the extra query is fine. + """ + result = await self.session.execute( + select(func.count(TaskTable.id)).where( + TaskTable.parent_task_id == task_id, + TaskTable.status.in_([TaskStatus.COMPLETED, TaskStatus.CANCELLED]), + ) + ) + return result.scalar_one() + async def self_heal_ac_ids(self, parent: TaskTable) -> None: """Re-stamp ``acceptance_criteria_ids`` in place when it's empty or out of length with ``acceptance_criteria`` -- a legacy row from before every @@ -10927,6 +10957,13 @@ class TaskService(BaseService): self._emit_admin_override_audit( task, pre_status, restored_status, actor_id, actor_role ) + # The oscillation-strikes marker is cleared by the caller + # (_admin_out_of_blocked) for every admin exit from BLOCKED, + # snapshot-driven or not — this branch only runs when a snapshot + # exists, so it must not clear it again here (the in-band + # unblock(restore=True) path never reaches this branch either — + # its own gateway call site reads/bumps this marker around this + # same restore). if ( restored_status == TaskStatus.NEEDS_REVISION and pre_status != TaskStatus.NEEDS_REVISION.value @@ -10980,10 +11017,16 @@ class TaskService(BaseService): pending/in_progress with a snapshot → full pre-block restore (returns the restored task). Review/queue targets → clear the stale claim so the next claimant starts clean (returns None; caller sets status). - Any other target → no-op (returns None). + Any other target → no-op (returns None). Every branch below also + clears the oscillation breaker's strike marker — a human is the one + moving the task out of BLOCKED here, so its job for this cycle is + done regardless of whether a pre-block snapshot survived to drive a + restore (a trip can wipe it before the force-block re-lands the task + in BLOCKED with no snapshot of its own). """ - if from_status != TaskStatus.BLOCKED.value: + if from_status != TaskStatus.BLOCKED.value or new_status == TaskStatus.BLOCKED: return None + markers.clear_marker(task, markers.OSCILLATION_STRIKES) if ( new_status in (TaskStatus.PENDING, TaskStatus.IN_PROGRESS) and task.pre_block_assignee is not None diff --git a/tests/unit/api/test_schemas_tasks.py b/tests/unit/api/test_schemas_tasks.py index 6577d7e8..6b0a046e 100644 --- a/tests/unit/api/test_schemas_tasks.py +++ b/tests/unit/api/test_schemas_tasks.py @@ -32,8 +32,17 @@ from roboco.api.schemas.tasks import ( task_to_response, transform_update_data, ) -from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team +from roboco.db.tables import TaskTable +from roboco.models.base import ( + BlockerResolverType, + Complexity, + TaskNature, + TaskStatus, + TaskType, + Team, +) from roboco.models.product import ProductCellMapping +from roboco.runtime.orchestrator import AgentOrchestrator _ORDER_DEFAULT = 0 @@ -311,6 +320,7 @@ def _stub_task(*, with_project: bool = False) -> Any: description="d", acceptance_criteria=["a"], status=TaskStatus.PENDING, + blocker_resolver_type=None, priority=1, sequence=0, nature=TaskNature.TECHNICAL, @@ -488,6 +498,90 @@ def test_task_list_to_response_returns_list() -> None: assert len(out) == len(stubs) +# --------------------------------------------------------------------------- +# blocker_resolver_type — wire-shaped regression. A hand-rolled stub (like +# _stub_task above) can't catch a field TaskResponse silently drops; this +# builds a REAL TaskTable row and pushes it through the actual serialization +# path the orchestrator's dispatchers see over HTTP. +# --------------------------------------------------------------------------- + + +def _hitl_blocked_task_row() -> TaskTable: + """A real TaskTable instance (never added to a session) in the exact + shape `unblock`'s oscillation-breaker trip leaves behind: BLOCKED with + blocker_resolver_type=HUMAN.""" + return TaskTable( + id=uuid4(), + title="t", + description="d", + acceptance_criteria=["a"], + acceptance_criteria_ids=[], + parent_ac_refs=[], + status=TaskStatus.BLOCKED, + blocker_resolver_type=BlockerResolverType.HUMAN, + priority=2, + sequence=0, + nature=TaskNature.TECHNICAL, + task_type=TaskType.CODE, + project_id=None, + product_id=None, + docs_complete=False, + pr_created=False, + board_review_complete=False, + team=Team.BACKEND, + created_by=uuid4(), + assigned_to=None, + parent_task_id=None, + dependency_ids=[], + blocker_ids=[], + batch_id=None, + created_at=datetime.now(UTC), + updated_at=None, + claimed_at=None, + claimed_by=None, + started_at=None, + completed_at=None, + target_date=None, + estimated_complexity=Complexity.LOW, + plan=None, + checkpoints=[], + progress_updates=[], + commits=[], + documents=[], + dev_notes=None, + qa_notes=None, + auditor_notes=None, + self_verified=False, + qa_verified=None, + branch_name=None, + pr_number=None, + pr_url=None, + source="manual", + confirmed_by_human=True, + ) + + +def test_task_to_response_serializes_blocker_resolver_type() -> None: + """A real ORM row's blocker_resolver_type must round-trip — the field was + silently missing from TaskResponse, so every dispatcher reading it over + the wire saw None regardless of the DB value.""" + row = _hitl_blocked_task_row() + resp = task_to_response(row) + assert resp.blocker_resolver_type == BlockerResolverType.HUMAN + + +def test_wire_shaped_hitl_blocked_task_trips_is_hitl_blocked() -> None: + """End-to-end wire simulation: real TaskTable -> task_to_response -> + JSON-mode serialization (what httpx.json() hands the orchestrator) -> + AgentOrchestrator._is_hitl_blocked. Before the fix this always read + None over the wire and never fired.""" + row = _hitl_blocked_task_row() + resp = task_to_response(row) + wire_dict = resp.model_dump(mode="json") + assert wire_dict["blocker_resolver_type"] == "human" + assert AgentOrchestrator._is_hitl_blocked(wire_dict) is True + + # --------------------------------------------------------------------------- # enrich_task_with_context — covers the work_session + project lookup branches. # --------------------------------------------------------------------------- diff --git a/tests/unit/foundation/policy/content/test_markers.py b/tests/unit/foundation/policy/content/test_markers.py index cfda46e9..f920fe25 100644 --- a/tests/unit/foundation/policy/content/test_markers.py +++ b/tests/unit/foundation/policy/content/test_markers.py @@ -8,6 +8,7 @@ from roboco.foundation.policy.content import markers as m # Named constant — ruff PLR2004 forbids magic-value comparisons. _TWO = 2 +_THREE = 3 def _task(om: dict | None = None) -> SimpleNamespace: @@ -155,3 +156,47 @@ def test_block_flip_count_bump_and_notify() -> None: assert m.is_block_flip_notified(t) is True # Marking notified must not reset the counter. assert m.get_block_flip_count(t) == _TWO + + +def test_oscillation_strikes_accrue_on_unchanged_fingerprint() -> None: + t = _task() + assert m.get_oscillation_strikes(t) == 0 + assert m.is_oscillation_tripped(t) is False + assert m.bump_oscillation_strikes(t, [0, 0]) == 1 + assert m.bump_oscillation_strikes(t, [0, 0]) == _TWO + assert m.bump_oscillation_strikes(t, [0, 0]) == _THREE + assert m.get_oscillation_strikes(t) == _THREE + assert m.is_oscillation_tripped(t) is False + + +def test_oscillation_strikes_reset_on_progress() -> None: + t = _task() + m.bump_oscillation_strikes(t, [0, 0]) + assert m.bump_oscillation_strikes(t, [0, 0]) == _TWO + # A new commit landed between rounds — real progress resets to 1. + assert m.bump_oscillation_strikes(t, [1, 0]) == 1 + # A revision round completing is progress too. + assert m.bump_oscillation_strikes(t, [1, 0]) == _TWO + assert m.bump_oscillation_strikes(t, [1, 1]) == 1 + + +def test_mark_oscillation_tripped_preserves_strikes_and_fingerprint() -> None: + t = _task() + m.bump_oscillation_strikes(t, [2, 1]) + m.bump_oscillation_strikes(t, [2, 1]) + m.mark_oscillation_tripped(t) + assert m.is_oscillation_tripped(t) is True + assert m.get_oscillation_strikes(t) == _TWO + # tripped survives a subsequent bump (belt-and-suspenders — the guard is + # meant to refuse before another bump ever happens). + m.bump_oscillation_strikes(t, [2, 1]) + assert m.is_oscillation_tripped(t) is True + + +def test_clear_marker_removes_oscillation_state() -> None: + t = _task() + m.bump_oscillation_strikes(t, [0, 0]) + m.mark_oscillation_tripped(t) + m.clear_marker(t, m.OSCILLATION_STRIKES) + assert m.get_oscillation_strikes(t) == 0 + assert m.is_oscillation_tripped(t) is False diff --git a/tests/unit/gateway/test_oscillation_breaker.py b/tests/unit/gateway/test_oscillation_breaker.py new file mode 100644 index 00000000..38441b01 --- /dev/null +++ b/tests/unit/gateway/test_oscillation_breaker.py @@ -0,0 +1,222 @@ +"""The escalate_up/unblock oscillation breaker. + +Live wedge: a cell PM's escalate_up auto-blocks a task and the Main PM's +unblock restores it, repeat, forever — the per-(agent, task) respawn breaker +in the orchestrator misses this because the escalator and the resolver each +own only half the round trips, so neither individual counter accrues at the +cycle's real rate. ``unblock`` now stamps a task-scoped, progress- +discriminated strike counter (``markers.oscillation_strikes``) on every +restore; past the trip threshold the task is force-blocked for a human +instead of restored again, and further ``unblock`` calls on it are refused +until an admin override clears it. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.foundation.policy.content import markers +from roboco.models.base import BlockerResolverType, TaskStatus +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + +# Mirrors _OSCILLATION_TRIP_THRESHOLD in _impl.py — ruff PLR2004 forbids +# magic-value comparisons. +_TRIP_THRESHOLD = 5 + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + base["journal"].has_decision_for_task.return_value = True + base["journal"].latest_decision_at.return_value = datetime.now(UTC) + return ChoreographerDeps(**base) + + +def _osc_setup() -> tuple[Choreographer, Any, Any, Any, Any]: + """A blocked task whose ``unblock_with_restore`` returns the SAME mock + object each call (one real ORM row across requests), with commits/ + revision_count seeded so the progress fingerprint is well-formed. + """ + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + status="blocked", + pre_block_state="in_progress", + pre_block_assignee=uuid4(), + pre_block_metadata={}, + dependency_ids=[], + orchestration_markers=None, + commits=[], + revision_count=0, + blocker_raised_by=uuid4(), + ) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.unblock_with_restore.return_value = t + task_svc.unmet_dependency_ids.return_value = [] + # A PM coordination root never commits itself — held constant here so + # existing scenarios (driven purely by commits/revision_count) are + # unaffected; tests targeting the child-count component override it. + task_svc.terminal_children_count = AsyncMock(return_value=0) + c = Choreographer(_make_deps(task=task_svc)) + return c, pm_id, task_id, t, task_svc + + +async def _unblock_once(c: Choreographer, pm_id: Any, task_id: Any, t: Any) -> Any: + """Re-block before each call — a fresh round trip in the cycle.""" + t.status = "blocked" + return await c.unblock(pm_id, task_id, "resolved upstream; restoring") + + +@pytest.mark.asyncio +async def test_no_progress_trips_after_threshold_cycles() -> None: + c, pm_id, task_id, t, task_svc = _osc_setup() + cc: Any = c + notify = AsyncMock() + cc._notify_ceo_oscillation = notify + + envs = [ + await _unblock_once(c, pm_id, task_id, t) for _ in range(_TRIP_THRESHOLD + 1) + ] + + for env in envs[:-1]: + assert env.error is None, env.as_dict() + tripped = envs[-1] + assert tripped.error is None, tripped.as_dict() + assert "force-blocked" in tripped.next + assert markers.is_oscillation_tripped(t) is True + assert markers.get_oscillation_strikes(t) == _TRIP_THRESHOLD + 1 + notify.assert_awaited_once() + assert t.blocker_resolver_type == BlockerResolverType.HUMAN + task_svc.admin_set_status.assert_awaited_once_with( + task_id, TaskStatus.BLOCKED, actor_role="system" + ) + + +@pytest.mark.asyncio +async def test_tripped_task_refuses_further_unblock() -> None: + c, pm_id, task_id, t, task_svc = _osc_setup() + cc: Any = c + cc._notify_ceo_oscillation = AsyncMock() + markers.mark_oscillation_tripped(t) + admin_calls_before = task_svc.admin_set_status.await_count + + env = await _unblock_once(c, pm_id, task_id, t) + + assert env.error == "invalid_state" + assert "force-blocked" in env.message + # The guard short-circuits before the restore path ever runs again. + assert task_svc.admin_set_status.await_count == admin_calls_before + task_svc.unblock_with_restore.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_progress_between_rounds_prevents_trip() -> None: + """A commit landing between escalations is real progress, not a loop — + strikes reset every round, so the breaker never trips no matter how many + rounds occur.""" + c, pm_id, task_id, t, _task_svc = _osc_setup() + cc: Any = c + cc._notify_ceo_oscillation = AsyncMock() + + for i in range(_TRIP_THRESHOLD + 3): + # A fresh commit lands every round — the commit count strictly grows. + t.commits = [{"sha": str(j)} for j in range(i + 1)] + env = await _unblock_once(c, pm_id, task_id, t) + assert env.error is None, env.as_dict() + + assert markers.is_oscillation_tripped(t) is False + assert markers.get_oscillation_strikes(t) == 1 + cc._notify_ceo_oscillation.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_revision_count_advance_also_counts_as_progress() -> None: + c, pm_id, task_id, t, _task_svc = _osc_setup() + cc: Any = c + cc._notify_ceo_oscillation = AsyncMock() + + for _ in range(_TRIP_THRESHOLD): + t.revision_count += 1 # a genuine revision round resolved + env = await _unblock_once(c, pm_id, task_id, t) + assert env.error is None, env.as_dict() + + assert markers.is_oscillation_tripped(t) is False + cc._notify_ceo_oscillation.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_static_terminal_children_still_trips() -> None: + """A coordination root with existing terminal children that DON'T change + between rounds still trips — the child count is one more progress + signal, not blanket forgiveness for an otherwise stalled cycle.""" + c, pm_id, task_id, t, task_svc = _osc_setup() + cc: Any = c + cc._notify_ceo_oscillation = AsyncMock() + task_svc.terminal_children_count = AsyncMock(return_value=3) + + envs = [ + await _unblock_once(c, pm_id, task_id, t) for _ in range(_TRIP_THRESHOLD + 1) + ] + + tripped = envs[-1] + assert tripped.error is None, tripped.as_dict() + assert markers.is_oscillation_tripped(t) is True + cc._notify_ceo_oscillation.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_terminal_child_completing_between_rounds_resets_strikes() -> None: + """On a PM coordination root the commit/revision_count components are + structurally static (PMs never commit) — a child landing COMPLETED + between escalations is the only progress signal available, and must + still reset strikes like the other two.""" + c, pm_id, task_id, t, task_svc = _osc_setup() + cc: Any = c + cc._notify_ceo_oscillation = AsyncMock() + + counts = iter([0, 0, 1, 1, 2]) # a child completes on round 3, then round 5 + task_svc.terminal_children_count = AsyncMock(side_effect=lambda _tid: next(counts)) + + for _ in range(5): + env = await _unblock_once(c, pm_id, task_id, t) + assert env.error is None, env.as_dict() + + assert markers.is_oscillation_tripped(t) is False + # Round 5's fingerprint differs from round 4's (1 -> 2 children), so this + # round's strike count reset to 1. + assert markers.get_oscillation_strikes(t) == 1 + cc._notify_ceo_oscillation.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_notify_named_with_strikes_and_both_agents() -> None: + c, pm_id, task_id, t, _task_svc = _osc_setup() + cc: Any = c + notify = AsyncMock() + cc._notify_ceo_oscillation = notify + escalator_id = t.blocker_raised_by + + for _ in range(_TRIP_THRESHOLD + 1): + await _unblock_once(c, pm_id, task_id, t) + + notify.assert_awaited_once_with( + task_id, + _TRIP_THRESHOLD + 1, + t.title, + escalator_id=escalator_id, + resolver_id=pm_id, + ) diff --git a/tests/unit/runtime/test_blocker_and_claimed_dispatch.py b/tests/unit/runtime/test_blocker_and_claimed_dispatch.py index 74a0ba22..a40cd20c 100644 --- a/tests/unit/runtime/test_blocker_and_claimed_dispatch.py +++ b/tests/unit/runtime/test_blocker_and_claimed_dispatch.py @@ -99,6 +99,63 @@ def test_blocked_task_non_cell_team_unassigned_is_unroutable() -> None: assert orch._blocker_resolver_slug(task) is None +# --------------------------------------------------------------------------- +# _dispatch_blocker_work — wire-shaped HITL skip. `_fetch_tasks` hands this +# dispatcher plain dicts decoded straight from GET /tasks JSON — this pins +# that shape (blocker_resolver_type as the lowercase enum-value string +# TaskResponse now serializes) rather than an in-process TaskTable/enum. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dispatch_blocker_work_skips_wire_shaped_hitl_blocked_task( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = _orch() + task: dict[str, Any] = { + "id": "t1", + "status": "blocked", + "blocker_resolver_type": "human", + "team": "backend", + "assigned_to": AGENT_UUIDS["be-dev-1"], + } + monkeypatch.setattr(orch, "_fetch_tasks", AsyncMock(return_value=[task])) + spawn = AsyncMock() + monkeypatch.setattr(orch, "spawn_agent", spawn) + + await orch._dispatch_blocker_work(client=MagicMock()) + + spawn.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_dispatch_blocker_work_spawns_non_hitl_blocked_task( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Control case: an agent-resolvable block (no HITL marker) still + dispatches normally — the wire-shaped skip above isn't just refusing + every blocked task.""" + orch = _orch() + task: dict[str, Any] = { + "id": "t1", + "status": "blocked", + "blocker_resolver_type": None, + "team": "backend", + "assigned_to": AGENT_UUIDS["be-pm"], + } + monkeypatch.setattr(orch, "_fetch_tasks", AsyncMock(return_value=[task])) + monkeypatch.setattr(orch, "_is_agent_active", lambda _agent_id: False) + monkeypatch.setattr(orch, "_pm_respawn_should_gate", AsyncMock(return_value=False)) + monkeypatch.setattr(orch, "_build_pm_blocker_prompt", lambda _task: "p") + monkeypatch.setattr(orch, "_task_git_context", lambda _task: None) + spawn = AsyncMock() + monkeypatch.setattr(orch, "spawn_agent", spawn) + + await orch._dispatch_blocker_work(client=MagicMock()) + + spawn.assert_awaited_once() + + # --------------------------------------------------------------------------- # _claimed_task_needs_agent — claimed-but-no-agent detection # --------------------------------------------------------------------------- @@ -318,6 +375,43 @@ async def test_dispatch_claimed_without_agent_spawns_at_most_one_per_tick( spawn.assert_awaited_once() +@pytest.mark.asyncio +async def test_dispatch_claimed_without_agent_has_no_progress_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unlike `_dispatch_blocker_work` (every spawn gated through + `_pm_respawn_should_gate`), this dispatcher carries no respawn-loop + protection of its own — it unconditionally respawns an agentless + claimed/in_progress task every tick past the grace window. This is half + of why the escalate_up/unblock oscillation defeats the per-(agent, task) + breaker: the resolved side of the cycle (the PM restored to in_progress) + has no counter here to ever trip, so the loop's other half never runs out + of fuel on its own — only a task-scoped breaker that also covers this + path (by moving the task to `blocked` entirely, which this dispatcher + doesn't fetch) can stop it. + """ + orch = _orch() + task = { + "id": "t1", + "status": "in_progress", + "assigned_to": AGENT_UUIDS["fe-pm"], + "updated_at": _STALE, + } + monkeypatch.setattr(orch, "_fetch_tasks", AsyncMock(return_value=[task])) + _stub_git_context(orch, monkeypatch) + monkeypatch.setattr(orch, "_get_prompt_for_agent", AsyncMock(return_value="p")) + spawn = AsyncMock() + monkeypatch.setattr(orch, "spawn_agent", spawn) + + cycles = 10 + for _ in range(cycles): + orch._tick_handled_tasks = set() # a fresh dispatch tick each cycle + await orch._dispatch_claimed_without_agent(client=MagicMock()) + + # No cycle was ever refused — zero backoff anywhere in this call path. + assert spawn.await_count == cycles + + @pytest.mark.asyncio async def test_dispatch_claimed_without_agent_releases_unknown_without_spending_budget( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/services/test_task.py b/tests/unit/services/test_task.py index d132a7cc..3c232a75 100644 --- a/tests/unit/services/test_task.py +++ b/tests/unit/services/test_task.py @@ -530,6 +530,58 @@ async def test_unblock_notifies_once_not_twice_on_repeated_call() -> None: mock_ns.send_unblock_notification.assert_awaited_once() +@pytest.mark.asyncio +async def test_unblock_clears_tripped_oscillation_marker() -> None: + """The legacy human/panel unblock route never goes through the gateway's + _oscillation_unblock_guard — reaching a tripped task here IS the human + intervention the breaker demands, so it must clear the marker rather + than leave it to refuse the task's next legitimate cycle.""" + task = _build_task( + status=TaskStatus.BLOCKED, branch_name=None, blocker_raised_by=uuid4() + ) + for _ in range(6): + markers.bump_oscillation_strikes(task, [0, 0, 0]) + markers.mark_oscillation_tripped(task) + assert markers.is_oscillation_tripped(task) is True + + svc = TaskService(MagicMock(flush=AsyncMock())) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_index_lifecycle_event_background", AsyncMock()) + mock_ns = MagicMock() + mock_ns.send_unblock_notification = AsyncMock() + with patch( + "roboco.services.notification.NotificationService", return_value=mock_ns + ): + out = await svc.unblock(task.id) + + assert out is task + assert markers.is_oscillation_tripped(task) is False + assert markers.get_oscillation_strikes(task) == 0 + + +@pytest.mark.asyncio +async def test_unblock_leaves_untripped_marker_alone() -> None: + """A normal (non-tripped) unblock must not reset an in-flight, still-live + strike count — only a tripped marker is cleared.""" + task = _build_task( + status=TaskStatus.BLOCKED, branch_name=None, blocker_raised_by=uuid4() + ) + markers.bump_oscillation_strikes(task, [0, 0, 0]) + assert markers.is_oscillation_tripped(task) is False + + svc = TaskService(MagicMock(flush=AsyncMock())) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_index_lifecycle_event_background", AsyncMock()) + mock_ns = MagicMock() + mock_ns.send_unblock_notification = AsyncMock() + with patch( + "roboco.services.notification.NotificationService", return_value=mock_ns + ): + await svc.unblock(task.id) + + assert markers.get_oscillation_strikes(task) == 1 + + @pytest.mark.asyncio async def test_wire_sibling_collision_dag_notifies_only_for_new_edges() -> None: """Collision-sequencing notification fires only for freshly-added edges. @@ -2207,6 +2259,101 @@ async def test_admin_set_status_force_no_revision_bump( ) +# --------------------------------------------------------------------------- +# Oscillation breaker — post-trip topology. The trip fires AFTER the restore +# already wiped pre_block_assignee/pre_block_state (a fresh re-block via +# admin_set_status stamps no new snapshot), so a CEO override out of BLOCKED +# must clear the marker even with no snapshot to drive a restore — otherwise +# it latently refuses this task's next legitimate gateway unblock forever. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_ceo_override_clears_oscillation_marker_with_wiped_snapshot( + db_session: AsyncSession, +) -> None: + agent = AgentTable( + id=uuid4(), + name="A", + slug=f"a-{uuid4().hex[:8]}", + role=AgentRole.MAIN_PM, + team=Team.BACKEND, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="pm", + capabilities=[], + permissions={}, + metrics={}, + ) + db_session.add(agent) + await db_session.flush() + project = ProjectTable( + id=uuid4(), + name="P", + slug=f"p-{uuid4().hex[:6]}", + git_url="https://example.com/r.git", + assigned_cell=Team.BACKEND, + created_by=agent.id, + ) + db_session.add(project) + await db_session.flush() + tid = uuid4() + task = TaskTable( + id=tid, + title="t", + description="d", + acceptance_criteria=["done"], + status=TaskStatus.BLOCKED, + priority=2, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.LOW, + team=Team.BACKEND, + confirmed_by_human=True, + project_id=project.id, + created_by=agent.id, + assigned_to=agent.id, + branch_name="feature/x", + # The exact post-trip topology: unblock_with_restore already wiped + # the snapshot before the trip fired, and the force-block into + # BLOCKED (admin_set_status, from a non-BLOCKED from_status) stamps + # no new one of its own. + pre_block_state=None, + pre_block_assignee=None, + blocker_resolver_type=BlockerResolverType.HUMAN, + ) + db_session.add(task) + await db_session.flush() + # A real trip: strikes accrued past threshold with an unchanging + # fingerprint, then force-blocked. + for _ in range(6): + markers.bump_oscillation_strikes(task, [0, 0, 0]) + markers.mark_oscillation_tripped(task) + assert markers.is_oscillation_tripped(task) is True + await db_session.flush() + + svc = get_task_service(db_session) + out = await svc.admin_set_status( + tid, TaskStatus.IN_PROGRESS, actor_id=cast("UUID", agent.id), actor_role="ceo" + ) + assert out is not None + await db_session.flush() + + row = ( + await db_session.execute(select(TaskTable).where(TaskTable.id == tid)) + ).scalar_one() + assert row.status == TaskStatus.IN_PROGRESS + assert markers.is_oscillation_tripped(row) is False + assert markers.get_oscillation_strikes(row) == 0, ( + "the marker must be gone entirely, not just its tripped flag, so the " + "next cycle's first bump starts a fresh strike count" + ) + # A later legitimate block/unblock cycle counts from fresh, not from the + # old strike count. + fresh_strikes = markers.bump_oscillation_strikes(row, [1, 0, 0]) + assert fresh_strikes == 1 + + # --------------------------------------------------------------------------- # _extract_completion_learnings — dead-letter on record_learning failure # ---------------------------------------------------------------------------