diff --git a/roboco/agents_config.py b/roboco/agents_config.py index 37a7ad63..c6ff35bf 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -122,7 +122,7 @@ def _verify_expiring_token( except (ValueError, KeyError): return False exp = payload.get("exp") - if not isinstance(exp, (int, float)): + if not isinstance(exp, int | float): return False # Short-circuit like the original: time.time() is read only when the # (id, role, team) fields already match, never on every call. diff --git a/roboco/api/auth/routes.py b/roboco/api/auth/routes.py index a6b5c311..1caaed5d 100644 --- a/roboco/api/auth/routes.py +++ b/roboco/api/auth/routes.py @@ -66,7 +66,7 @@ async def revoke_and_logout( ) jti = data.get("jti") exp = data.get("exp") - if isinstance(jti, str) and isinstance(exp, (int, float)): + if isinstance(jti, str) and isinstance(exp, int | float): await revocation.revoke_jti(jti, max(int(exp - time.time()), 1)) except Exception: _logger.warning("logout jti revocation skipped", exc_info=True) diff --git a/roboco/api/deps.py b/roboco/api/deps.py index 596178e3..67fb57a6 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -477,7 +477,7 @@ def _should_remint(token: str) -> bool: }, ) exp = data.get("exp") - if isinstance(exp, (int, float)): + if isinstance(exp, int | float): return (exp - time.time()) < settings.cloud_auth_remint_threshold_seconds except Exception: return True diff --git a/roboco/config.py b/roboco/config.py index fbe3773b..c259ec9e 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -1726,6 +1726,33 @@ class Settings(BaseSettings): "the branch reached the remote." ), ) + evidence_assembly_timeout_seconds: float = Field( + default=45.0, + ge=1.0, + description=( + "TOTAL budget for one advisory-evidence build (branch fetch, " + "diff, list_changed_files — every slow leg) on claim_review / " + "claim_doc_task / claim_gate_review / evidence() / i_am_done's " + "success envelope. A `LegBudget` (roboco.services.gateway." + "choreographer.evidence_legs) is created once per build and " + "shared across every leg in it — each leg's wait_for gets only " + "what's left of this total, shrinking as legs consume it, so " + "summing per-leg budgets can never exceed this cap (and stays " + "well under flow_verb_timeout_seconds). A hung leg degrades " + "(evidence_gaps) instead of taking the whole verb down with it." + ), + ) + conventions_validator_advisory_timeout_seconds: float = Field( + default=30.0, + ge=1.0, + description=( + "Ceiling for the conventions-validator subprocess on the " + "ADVISORY claim path (claim_review) — the actual budget used is " + "min(this, the build's remaining evidence_assembly_timeout_seconds), " + "so it also shrinks with the shared LegBudget. Fail-closed paths " + "(i_am_done, pr_pass) keep their own hardcoded cap unchanged." + ), + ) protected_git_urls: list[str] = Field( default_factory=list, diff --git a/roboco/llm/providers/codex_auth.py b/roboco/llm/providers/codex_auth.py index d1b6e1e6..5ab9c982 100644 --- a/roboco/llm/providers/codex_auth.py +++ b/roboco/llm/providers/codex_auth.py @@ -108,7 +108,7 @@ def _exp_from_jwt(token: str) -> datetime | None: if not isinstance(payload, dict): return None exp = payload.get("exp") - if not isinstance(exp, (int, float)): + if not isinstance(exp, int | float): return None return datetime.fromtimestamp(float(exp), tz=UTC) diff --git a/roboco/llm/providers/codex_cli_usage.py b/roboco/llm/providers/codex_cli_usage.py index 364ae265..0d6b72b1 100644 --- a/roboco/llm/providers/codex_cli_usage.py +++ b/roboco/llm/providers/codex_cli_usage.py @@ -59,7 +59,7 @@ _USAGE_FIELDS = ( def _as_int(value: object) -> int: - return int(value) if isinstance(value, (int, float)) else 0 + return int(value) if isinstance(value, int | float) else 0 def _usage_from_event(event: dict[str, Any]) -> dict[str, int] | None: diff --git a/roboco/llm/providers/gemini_cli_usage.py b/roboco/llm/providers/gemini_cli_usage.py index a81ac25e..eea29647 100644 --- a/roboco/llm/providers/gemini_cli_usage.py +++ b/roboco/llm/providers/gemini_cli_usage.py @@ -95,7 +95,7 @@ _RATE_LIMIT_EXIT_CODE = 75 def _coerce_int(value: object) -> int: - return int(value) if isinstance(value, (int, float)) else 0 + return int(value) if isinstance(value, int | float) else 0 def _model_tokens(entry: dict[str, Any]) -> tuple[int, int]: diff --git a/roboco/llm/providers/grok_auth.py b/roboco/llm/providers/grok_auth.py index c2a891de..475117fb 100644 --- a/roboco/llm/providers/grok_auth.py +++ b/roboco/llm/providers/grok_auth.py @@ -202,7 +202,7 @@ def _exp_from_access_token(access_token: str) -> datetime | None: if not isinstance(payload, dict): return None exp = payload.get("exp") - if not isinstance(exp, (int, float)): + if not isinstance(exp, int | float): return None return datetime.fromtimestamp(float(exp), tz=UTC) @@ -215,7 +215,7 @@ def _apply_refreshed_token( if token.get("refresh_token"): creds["refresh_token"] = token["refresh_token"] expires_in = token.get("expires_in") - if isinstance(expires_in, (int, float)): + if isinstance(expires_in, int | float): creds["expires_at"] = _to_iso_z(now + timedelta(seconds=float(expires_in))) else: # xAI's refresh response sometimes omits expires_in. The access token diff --git a/roboco/llm/providers/grok_cli_usage.py b/roboco/llm/providers/grok_cli_usage.py index 25e3913e..46c98d33 100644 --- a/roboco/llm/providers/grok_cli_usage.py +++ b/roboco/llm/providers/grok_cli_usage.py @@ -86,9 +86,9 @@ def _extract_total_tokens(event: dict[str, Any]) -> int: for meta in (update_meta, params_meta): if isinstance(meta, dict) and "totalTokens" in meta: value = meta["totalTokens"] - return int(value) if isinstance(value, (int, float)) else 0 + return int(value) if isinstance(value, int | float) else 0 value = event.get("totalTokens", 0) - return int(value) if isinstance(value, (int, float)) else 0 + return int(value) if isinstance(value, int | float) else 0 def find_updates_path(grok_home: Path, cwd: str, session_id: str) -> Path | None: diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index ca8d4b64..7e1c4041 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -14,7 +14,7 @@ from __future__ import annotations import contextlib from dataclasses import dataclass from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any, ClassVar, cast +from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast from uuid import UUID import structlog @@ -33,6 +33,10 @@ from roboco.services.gateway.choreographer import findings as findings_lib from roboco.services.gateway.choreographer._protocol import actor_context_fields from roboco.services.gateway.choreographer._verb_runner import VerbRunner from roboco.services.gateway.choreographer.collision import build_collision_context +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) from roboco.services.gateway.claim_guards import ( already_active_guard, paused_tasks_guard, @@ -81,6 +85,17 @@ if TYPE_CHECKING: from roboco.models.base import TaskNature from roboco.services.gateway.choreographer._protocol import ChoreographerHelpers +# _ensure_pm_decision's result: "fresh" (a recent decision already satisfies +# the window, nothing written), "wrote" (recorded now), "transient_failure" +# (the write raised WITH a non-empty rationale in hand — a DB-contention +# lock-timeout, not a real absence), "absent" (no rationale AND the window +# check never ran — legacy default for every call site that hasn't been +# threaded through yet). Every PM-decision gate helper accepts this as an +# optional param so "transient_failure" can satisfy the gate for THIS call +# (the rationale is in the verb payload; the write is a convenience, not the +# substance) without changing every verb's own public signature. +PmDecisionOutcome = Literal["fresh", "wrote", "transient_failure", "absent"] + # Minimum character length enforced on rich_plan["approach"] by the PM # sub-tasks gate. Must match the Pydantic min_length on # IWillPlanRequest.approach. Raised 20→150: plans were vague @@ -3384,7 +3399,7 @@ class Choreographer: async def _ensure_pm_decision( self, agent_id: UUID, task_id: UUID, rationale: str | None - ) -> None: + ) -> PmDecisionOutcome: """Auto-record a journal:decision from a PM verb's own rationale. The write-then-gate pattern (mirrors i_am_blocked → write_struggle): @@ -3399,7 +3414,10 @@ class Choreographer: Idempotent within the decision window: when a fresh decision already satisfies the gate, no duplicate is written. Best-effort — a journal write failure is logged and swallowed so the verb falls through to - the normal gate (which rejects as before), never crashing the verb. + the normal gate — which the caller can now short-circuit to PASS + (see ``PmDecisionOutcome`` / the gate helpers below) instead of + laundering a transient DB hiccup into a rejection the PM's own + rationale already answers. Savepoint-guarded: the journal INSERT can lock-timeout on a concurrent claim holding the task row's FK share lock (live @@ -3413,12 +3431,19 @@ class Choreographer: below rolls back to, leaving the outer transaction — and every object this call didn't itself touch, e.g. the caller's ``t`` — exactly as usable as if the write had never been attempted. + + Returns the outcome so the caller can thread it into the gate that + runs right after: "fresh" / "wrote" both mean a decision genuinely + exists; "transient_failure" means the write raised but a rationale + WAS in hand (the caller may treat the gate as satisfied for this + call); "absent" means there was no rationale to record at all (the + gate should reject exactly as before). """ from roboco.config import settings as _settings text = (rationale or "").strip() if not text: - return + return "absent" try: async with self.task.session.begin_nested(): latest = await self.journal.latest_decision_at(agent_id, task_id) @@ -3427,19 +3452,27 @@ class Choreographer: latest is not None and (datetime.now(UTC) - latest).total_seconds() <= window ): - return + return "fresh" await self.journal.write_decision( agent_id=agent_id, task_id=task_id, content=text ) - except Exception as exc: # best-effort; gate rejects normally on failure + return "wrote" + except Exception as exc: # best-effort; caller's gate handles the fallout logger.warning( "auto-record pm decision failed", error=str(exc), task_id=str(task_id), ) + return "transient_failure" async def _check_pm_decision_required( - self, verb: str, agent_id: UUID, task_id: UUID, t: Any + self, + verb: str, + agent_id: UUID, + task_id: UUID, + t: Any, + *, + pm_decision_outcome: PmDecisionOutcome | None = None, ) -> Envelope | None: """Standard PM-verb tracing gate driven by VERB_REQUIREMENTS. @@ -3454,6 +3487,15 @@ class Choreographer: than ``settings.pm_decision_window_seconds``. Older decisions are treated as missing so PMs write a fresh decision around each decision point rather than once at task creation. + + ``pm_decision_outcome`` (from the caller's own + ``_ensure_pm_decision`` call, defaulting to ``None`` for legacy + parity) short-circuits the freshness check to satisfied when it is + ``"transient_failure"`` — the write lock-timed out under DB + contention, but the verb's own rationale WAS in hand and is the + substance the gate actually cares about; the write was only ever a + convenience. Any other value (or ``None``) leaves this check + byte-for-byte unchanged. """ from roboco.config import settings as _settings from roboco.foundation.policy import tracing as _tr @@ -3468,6 +3510,14 @@ class Choreographer: latest is not None and (datetime.now(UTC) - latest).total_seconds() <= window_seconds ) + gate_satisfied_by_rationale = pm_decision_outcome == "transient_failure" + if gate_satisfied_by_rationale: + logger.warning( + "pm decision auto-record failed transiently; gate satisfied " + "by verb rationale", + task_id=str(task_id), + verb=verb, + ) # QUICK_CONTEXT_MIN_CHARS applies only to ``delegate`` (its # required-set is the only one carrying it); the PM pre-writes the @@ -3475,7 +3525,7 @@ class Choreographer: # Setting the threshold here is inert for unblock / escalate, which do # not require it. ctx = _tr.GateContext( - journal_decision_present=fresh, + journal_decision_present=fresh or gate_satisfied_by_rationale, quick_context_min_chars=_settings.quick_context_min_chars, ) result = _tr.check_requirements( @@ -3488,7 +3538,12 @@ class Choreographer: return await self._build_tracing_gap(agent_id, task_id, result.missing, task=t) async def _check_complete_gates( - self, agent_id: UUID, task_id: UUID, notes: str + self, + agent_id: UUID, + task_id: UUID, + notes: str, + *, + pm_decision_outcome: PmDecisionOutcome | None = None, ) -> Envelope | None: """Tracing gate for cell-PM and main-PM ``complete`` verbs. @@ -3498,6 +3553,13 @@ class Choreographer: guards because that gate emits a richer remediation message listing the non-terminal subtasks; keeping it inline preserves that UX. + + ``pm_decision_outcome`` mirrors ``_check_pm_decision_required``'s + param: ``"transient_failure"`` (the caller's ``_ensure_pm_decision`` + write lock-timed out but a rationale WAS provided) satisfies both + the decision and reflect legs for this call — same substitution + already applied to a genuinely-written decision below — instead of + rejecting a DB hiccup as a missing decision. """ from types import SimpleNamespace @@ -3506,9 +3568,17 @@ class Choreographer: has_decision = await self.journal.has_decision_for_task(agent_id, task_id) has_reflect = await self.journal.has_reflect_for_task(agent_id, task_id) + gate_satisfied_by_rationale = pm_decision_outcome == "transient_failure" + if gate_satisfied_by_rationale: + logger.warning( + "pm decision auto-record failed transiently; gate satisfied " + "by verb rationale", + task_id=str(task_id), + verb="complete", + ) task_view = SimpleNamespace(notes=notes) ctx = _tr.GateContext( - journal_decision_present=has_decision, + journal_decision_present=has_decision or gate_satisfied_by_rationale, # A PM closing/submitting a task documents it in its *decision* note; # a separate *reflect* adds little for a coordination/review close and # is exactly the artifact weak-model PMs forget — looping on the @@ -3516,7 +3586,9 @@ class Choreographer: # decision as satisfying reflect for the PM complete/submit_up close; # the gate still requires a decision + substantive notes, so the close # stays documented — only the redundant second-artifact demand drops. - journal_reflect_present=has_reflect or has_decision, + journal_reflect_present=( + has_reflect or has_decision or gate_satisfied_by_rationale + ), notes_min_chars=getattr(_settings, "notes_min_chars", 20), ) result = _tr.check_requirements( @@ -3529,7 +3601,12 @@ class Choreographer: return await self._build_tracing_gap(agent_id, task_id, result.missing) async def _check_submit_up_gates( - self, agent_id: UUID, task_id: UUID, notes: str + self, + agent_id: UUID, + task_id: UUID, + notes: str, + *, + pm_decision_outcome: PmDecisionOutcome | None = None, ) -> Envelope | None: """Tracing gate for ``submit_up`` (cell PM bubble-up). @@ -3543,6 +3620,11 @@ class Choreographer: ``submit_root`` (both route through ``_submit_up_guard``), so ``open_finding_ids`` covers pr_gate/pm/ceo-origin findings on either a cell root or a Main-PM root. + + ``pm_decision_outcome`` mirrors ``_check_complete_gates``'s param — + ``"transient_failure"`` satisfies the decision + reflect legs for + this call instead of rejecting a DB-contention write hiccup as a + missing decision. """ from types import SimpleNamespace @@ -3552,9 +3634,17 @@ class Choreographer: has_decision = await self.journal.has_decision_for_task(agent_id, task_id) has_reflect = await self.journal.has_reflect_for_task(agent_id, task_id) open_finding_ids = await self._open_finding_ids(task_id) + gate_satisfied_by_rationale = pm_decision_outcome == "transient_failure" + if gate_satisfied_by_rationale: + logger.warning( + "pm decision auto-record failed transiently; gate satisfied " + "by verb rationale", + task_id=str(task_id), + verb="submit_up", + ) task_view = SimpleNamespace(notes=notes) ctx = _tr.GateContext( - journal_decision_present=has_decision, + journal_decision_present=has_decision or gate_satisfied_by_rationale, # A PM closing/submitting a task documents it in its *decision* note; # a separate *reflect* adds little for a coordination/review close and # is exactly the artifact weak-model PMs forget — looping on the @@ -3562,7 +3652,9 @@ class Choreographer: # decision as satisfying reflect for the PM complete/submit_up close; # the gate still requires a decision + substantive notes, so the close # stays documented — only the redundant second-artifact demand drops. - journal_reflect_present=has_reflect or has_decision, + journal_reflect_present=( + has_reflect or has_decision or gate_satisfied_by_rationale + ), notes_min_chars=getattr(_settings, "notes_min_chars", 20), open_finding_ids=open_finding_ids, ) @@ -3622,13 +3714,31 @@ class Choreographer: files_changed sourced from git (authoritative) so the i_am_done envelope shows the same file list QA / docs / PMs will see — independent of legacy ``add_files_modified`` plumbing. + + Runs strictly AFTER the composed transition already committed (the + caller already ran submit_verification/submit_qa) — this is purely + informational, not a gate — so the list_changed_files leg runs + bounded via ``run_bounded_leg``: a timeout must not strand the + dev's already-succeeded submit behind a hung response. """ + from roboco.config import settings as _settings + journal_highlights = await self.evidence_repo.journal_highlights_for_task( task_id ) files_changed: list[str] = [] + evidence_gaps: list[str] = [] if t.branch_name: - files_changed = await self.git.list_changed_files(branch_name=t.branch_name) + budget = LegBudget(_settings.evidence_assembly_timeout_seconds) + files_changed = await run_bounded_leg( + self.git.list_changed_files(branch_name=t.branch_name), + default=[], + budget=budget, + leg="files_changed", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, + ) # Normally empty here — i_am_done's FINDINGS_ADDRESSED gate already # required every open finding resolved before this point — but wired # in for consistency with every other evidence call site. @@ -3640,6 +3750,7 @@ class Choreographer: journal_highlights=journal_highlights, files_changed=files_changed, revision_findings=open_findings, + evidence_gaps=evidence_gaps, ) agent = await self.task.agent_for(agent_id) role = str(agent.role) if agent is not None else "developer" @@ -5486,7 +5597,7 @@ class Choreographer: # Write-then-gate: the delegated subtask's title + description (the # PM's own articulation of the work) is recorded as the # journal:decision the tracing guard below requires. - await self._ensure_pm_decision( + pm_decision_outcome = await self._ensure_pm_decision( pm_agent_id, parent_task_id, f"Delegating subtask '{inputs.title}': {inputs.description}", @@ -5496,7 +5607,12 @@ class Choreographer: # coercion + assignee-vs-task_type, parent-ownership/subtask-cap, and # (last) decomposition-coverage. guard = await self._delegate_post_spec_guards( - pm_agent_id, parent_task_id, parent, role_str, inputs + pm_agent_id, + parent_task_id, + parent, + role_str, + inputs, + pm_decision_outcome=pm_decision_outcome, ) if guard is not None: return await self._emit_rejection( @@ -5516,6 +5632,8 @@ class Choreographer: parent: Any, role_str: str, inputs: DelegateInputs, + *, + pm_decision_outcome: PmDecisionOutcome, ) -> Envelope | None: """``_delegate_extra_guards``, then the decomposition-coverage gate. @@ -5526,7 +5644,12 @@ class Choreographer: gate only ever fires on an otherwise-valid delegate. """ guard = await self._delegate_extra_guards( - pm_agent_id, parent_task_id, parent, role_str, inputs + pm_agent_id, + parent_task_id, + parent, + role_str, + inputs, + pm_decision_outcome=pm_decision_outcome, ) if guard is not None: return guard @@ -5608,6 +5731,8 @@ class Choreographer: parent: Any, role_str: str, inputs: DelegateInputs, + *, + pm_decision_outcome: PmDecisionOutcome, ) -> Envelope | None: """Delegate-specific guards the spec doesn't model. @@ -5621,7 +5746,11 @@ class Choreographer: """ # Pre-gateway PM.md required journal:decision before each delegate. if env := await self._check_pm_decision_required( - "delegate", pm_agent_id, parent_task_id, parent + "delegate", + pm_agent_id, + parent_task_id, + parent, + pm_decision_outcome=pm_decision_outcome, ): return env chain_error = self._validate_delegation_chain(role_str, inputs.assigned_to) @@ -6970,8 +7099,12 @@ class Choreographer: """ # Write-then-gate: the bubble-up rationale (notes, already validated # >= min by the ownership guard) becomes the journal:decision. - await self._ensure_pm_decision(pm_agent_id, task_id, notes) - if env := await self._check_submit_up_gates(pm_agent_id, task_id, notes): + pm_decision_outcome = await self._ensure_pm_decision( + pm_agent_id, task_id, notes + ) + if env := await self._check_submit_up_gates( + pm_agent_id, task_id, notes, pm_decision_outcome=pm_decision_outcome + ): return env if env := await self._subtasks_not_terminal_envelope( pm_agent_id, task_id, context_phrase="bubbling up" @@ -7197,9 +7330,11 @@ class Choreographer: # journal:decision the gate below requires, so a PM that didn't # pre-call note(scope='decision') doesn't stall in a tracing_gap # respawn loop. - await self._ensure_pm_decision(pm_agent_id, task_id, reason) + pm_decision_outcome = await self._ensure_pm_decision( + pm_agent_id, task_id, reason + ) if env := await self._check_pm_decision_required( - "unblock", pm_agent_id, task_id, t + "unblock", pm_agent_id, task_id, t, pm_decision_outcome=pm_decision_outcome ): return await self._emit_rejection( env.with_introspection(task=t, role=role), @@ -7456,8 +7591,12 @@ class Choreographer: ) # Write-then-gate: the PM's merge rationale (notes) becomes the # journal:decision the gate requires — no separate note() call needed. - await self._ensure_pm_decision(pm_agent_id, task_id, notes) - if env := await self._check_complete_gates(pm_agent_id, task_id, notes): + pm_decision_outcome = await self._ensure_pm_decision( + pm_agent_id, task_id, notes + ) + if env := await self._check_complete_gates( + pm_agent_id, task_id, notes, pm_decision_outcome=pm_decision_outcome + ): return env if env := ( await self._subtasks_not_terminal_envelope( @@ -8236,9 +8375,14 @@ class Choreographer: ) # Write-then-gate: the Main PM's root-close rationale (notes) becomes # the journal:decision the gate requires. - await self._ensure_pm_decision(main_pm_agent_id, root_task_id, notes) - if env := await self._check_complete_gates( + pm_decision_outcome = await self._ensure_pm_decision( main_pm_agent_id, root_task_id, notes + ) + if env := await self._check_complete_gates( + main_pm_agent_id, + root_task_id, + notes, + pm_decision_outcome=pm_decision_outcome, ): return env if env := ( @@ -8731,9 +8875,16 @@ class Choreographer: # Write-then-gate: the escalation reason becomes the journal:decision # the preflight gate requires. - await self._ensure_pm_decision(pm_agent_id, task_id, reason) + pm_decision_outcome = await self._ensure_pm_decision( + pm_agent_id, task_id, reason + ) preflight = await self._escalate_up_preflight( - pm_agent_id, t, me, briefing, role_str + pm_agent_id, + t, + me, + briefing, + role_str, + pm_decision_outcome=pm_decision_outcome, ) if preflight is not None: return await self._emit_rejection( @@ -8778,6 +8929,8 @@ class Choreographer: me: Any, briefing: dict[str, Any], role_str: str, + *, + pm_decision_outcome: PmDecisionOutcome, ) -> Envelope | None: """Verb-specific preflight gates for escalate_up. @@ -8789,7 +8942,11 @@ class Choreographer: ``VERB_REQUIREMENTS["escalate_up"]``. """ if env := await self._check_pm_decision_required( - "escalate_up", pm_agent_id, t.id, t + "escalate_up", + pm_agent_id, + t.id, + t, + pm_decision_outcome=pm_decision_outcome, ): return env.with_introspection(task=t, role=role_str) target_slug = me.escalation_target if me else None @@ -8909,9 +9066,13 @@ class Choreographer: # completed subtask covered — inert until coverage is declared). # Write-then-gate: the escalation reason becomes the journal:decision # the gate below requires. - await self._ensure_pm_decision(agent_id, task_id, reason) + pm_decision_outcome = await self._ensure_pm_decision(agent_id, task_id, reason) env = await self._check_pm_decision_required( - "escalate_to_ceo", agent_id, task_id, t + "escalate_to_ceo", + agent_id, + task_id, + t, + pm_decision_outcome=pm_decision_outcome, ) or await self._parent_acs_covered_envelope( agent_id, task_id, context_phrase="escalating to CEO" ) diff --git a/roboco/services/gateway/choreographer/doc.py b/roboco/services/gateway/choreographer/doc.py index 96fc1b6a..51dbdf96 100644 --- a/roboco/services/gateway/choreographer/doc.py +++ b/roboco/services/gateway/choreographer/doc.py @@ -45,6 +45,10 @@ from roboco.foundation.policy import tracing as _tr from roboco.foundation.policy.content import markers from roboco.models.task import DocRef from roboco.services.gateway.choreographer import findings as findings_lib +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) from roboco.services.gateway.envelope import Envelope from roboco.services.gateway.evidence_builder import build_evidence_for_task @@ -196,13 +200,29 @@ class DocMixin(_Base): # the task branch already exists (dev created it) so no checkout # ran in the doc's workspace. Put the doc on the task branch now # so roboco_docs_write / commit don't fail BRANCH_MISMATCH. - # Best-effort — a checkout hiccup must not fail the claim. + # Best-effort — a checkout hiccup must not fail the claim. Bounded + # against the SAME shared LegBudget the evidence legs below use + # (one budget per build), so this leg + diff + list_changed_files + # can never sum past the total evidence-assembly budget. + budget = LegBudget(settings.evidence_assembly_timeout_seconds) + gaps: list[str] = [] if t.branch_name: with contextlib.suppress(Exception): - await self.git.checkout_branch_in_agent_workspace( - t.branch_name, actor_agent_id=doc_agent_id + await run_bounded_leg( + self.git.checkout_branch_in_agent_workspace( + t.branch_name, actor_agent_id=doc_agent_id + ), + default=None, + budget=budget, + leg="workspace checkout", + hint=( + "the diff below may reflect a stale worktree; " + "re-run roboco_git_diff to confirm" + ), + task_id=task_id, + gaps=gaps, ) - ev = await self._claim_doc_evidence(t, task_id) + ev = await self._claim_doc_evidence(t, task_id, budget=budget, extra_gaps=gaps) return Envelope.ok( status=str(t.status), task_id=str(task_id), @@ -211,19 +231,49 @@ class DocMixin(_Base): context_briefing=briefing, ).with_introspection(task=t, role=role_str) - async def _claim_doc_evidence(self, task: Any, task_id: UUID) -> dict[str, Any]: + async def _claim_doc_evidence( + self, + task: Any, + task_id: UUID, + *, + budget: LegBudget, + extra_gaps: list[str] | None = None, + ) -> dict[str, Any]: """Build the evidence dict surfaced inline on claim_doc_task ok envelopes. files_changed sourced from git (authoritative) instead of ``work_session.files_modified``, which the gateway commit() does not populate. The docs writer sees an accurate file list. + + The diff/list_changed_files legs run bounded via ``run_bounded_leg`` + against ``budget`` — the SAME shared instance the caller's checkout + leg already drew from, so a timeout skips that piece and records a + note in ``evidence_gaps`` instead of hanging this advisory verb, and + the whole build (checkout + diff + list_changed_files) stays under + one total. ``extra_gaps`` folds in the caller's own pre-evidence + gaps (the checkout leg above). """ files_changed: list[str] = [] diff = "" + evidence_gaps: list[str] = list(extra_gaps or []) if task.branch_name: - diff = await self.git.diff(branch_name=task.branch_name) - files_changed = await self.git.list_changed_files( - branch_name=task.branch_name + diff = await run_bounded_leg( + self.git.diff(branch_name=task.branch_name), + default="", + budget=budget, + leg="pr diff", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, + ) + files_changed = await run_bounded_leg( + self.git.list_changed_files(branch_name=task.branch_name), + default=[], + budget=budget, + leg="files_changed", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, ) journal_highlights = await self.evidence_repo.journal_highlights_for_task( task_id @@ -237,6 +287,7 @@ class DocMixin(_Base): files_changed=files_changed, pr_diff_summary=diff, revision_findings=open_findings, + evidence_gaps=evidence_gaps, ).as_dict() async def _verify_doc_owner( diff --git a/roboco/services/gateway/choreographer/evidence_legs.py b/roboco/services/gateway/choreographer/evidence_legs.py new file mode 100644 index 00000000..afd633b2 --- /dev/null +++ b/roboco/services/gateway/choreographer/evidence_legs.py @@ -0,0 +1,133 @@ +"""Bounded advisory-evidence legs for claim_review / claim_doc_task / +claim_gate_review / evidence() / i_am_done's success envelope. + +Claim-time (and i_am_done's post-transition) evidence assembly is advisory, +not gating: a slow branch fetch, diff, or list_changed_files must degrade +the evidence rather than hold the whole flow-verb request (and its +transaction's row locks) hostage for the full ``flow_verb_timeout_seconds`` +budget. ``run_bounded_leg`` wraps one such awaitable in ``asyncio.wait_for``; +on a timeout it appends a human-readable entry to the caller's +``evidence_gaps`` list, logs one structured warning, and returns the +caller's ``default`` instead of propagating. Fail-closed paths (``i_am_done`` +/ ``pr_pass`` conventions enforcement) do not use this — they keep their +existing hard cap. + +Two distinct timeout shapes get caught, because they fire from different +layers: + +- ``TimeoutError`` — asyncio's own cancellation-converted timeout, raised by + ``asyncio.wait_for`` itself when ``coro`` is still running at the deadline + (e.g. a slow DB read, a lock wait, or a git op that outlives its own + internal bound). +- ``GitTimeoutError`` — ``GitService._run_git``'s own internal subprocess + bound (``settings.git_command_timeout_seconds``, 30s by default — usually + SHORTER than a leg's own budget, making this the most common real-world + timeout shape for a single hung git call). It is a ``GitError`` / + ``RobocoError`` subclass, NOT a ``TimeoutError`` subclass, and is raised + from INSIDE the coroutine (the subprocess itself gave up), not by + ``wait_for``'s cancellation. Any OTHER ``GitError`` (a real command + failure — bad ref, auth, network refusal, not a timeout) still propagates + uncaught; only a timeout degrades. + +The conventions-validator subprocess leg does NOT go through +``run_bounded_leg`` — see ``QAMixin._qa_convention_findings``'s docstring +for why nesting an outer wait_for around a coroutine with its own inner +wait_for-based cleanup (``proc.kill()`` / ``proc.wait()``) can orphan the +child process. +""" + +from __future__ import annotations + +import asyncio +import time +from typing import TYPE_CHECKING, Any + +import structlog + +from roboco.exceptions import GitTimeoutError + +if TYPE_CHECKING: + from collections.abc import Coroutine + +logger = structlog.get_logger() + +# Floor so a nearly-exhausted budget still gives the next leg a real chance +# to run (and a real wait_for timeout > 0) instead of skipping it outright. +_MIN_LEG_SECONDS = 1.0 + + +class LegBudget: + """A shared, once-per-evidence-build deadline for a sequence of legs. + + Per-leg budgets summed naively can exceed the flow-verb wall (e.g. three + 45s legs = 135s, well past the 120s verb timeout). One ``LegBudget`` + instance per evidence build gives each leg only what's left of the + TOTAL assembly budget — shrinking as legs consume it, never resetting + per leg — so the whole build's wall time is capped near the configured + total regardless of how many legs it runs. + """ + + __slots__ = ("_deadline",) + + def __init__(self, total_seconds: float) -> None: + self._deadline = time.monotonic() + total_seconds + + def remaining(self) -> float: + """Seconds left before the deadline, floored at ``_MIN_LEG_SECONDS``.""" + return max(_MIN_LEG_SECONDS, self._deadline - time.monotonic()) + + +async def run_bounded_leg[T]( + coro: Coroutine[Any, Any, T], + *, + default: T, + budget: LegBudget, + leg: str, + hint: str, + task_id: Any, + gaps: list[str], +) -> T: + """Await ``coro`` bounded by ``budget``'s remaining time. + + Degrades to ``default`` and appends one entry to ``gaps`` on either + timeout shape (see module docstring); the entry names which bound + tripped so a reader can tell a slow-but-alive leg (the assembly budget) + from a git subprocess that gave up on its own shorter bound. + + Cancelling a ``run_in_executor``/``asyncio.to_thread``-backed git call + abandons the worker thread — this only stops AWAITING it, not the + underlying subprocess. Each git-touching leg's own subprocess call is + independently bounded (``git_command_timeout_seconds`` inside + ``_run_git``, or an explicit ``subprocess_timeout`` on + ``fetch_branch_for_inspection``) so the thread still self-terminates + near its own window rather than running to completion unbounded. + """ + timeout = budget.remaining() + try: + return await asyncio.wait_for(coro, timeout=timeout) + except TimeoutError: + gaps.append( + f"{leg} unavailable: timed out after {timeout:.0f}s " + f"(evidence-assembly budget) — {hint}" + ) + logger.warning( + "evidence_leg_timeout", + leg=leg, + timeout=timeout, + task_id=str(task_id), + bound="assembly_budget", + ) + return default + except GitTimeoutError as exc: + gaps.append( + f"{leg} unavailable: a git command timed out after {exc.timeout}s " + f"(git_command_timeout_seconds) — {hint}" + ) + logger.warning( + "evidence_leg_timeout", + leg=leg, + timeout=exc.timeout, + task_id=str(task_id), + bound="git_command_timeout", + ) + return default diff --git a/roboco/services/gateway/choreographer/pr_gate.py b/roboco/services/gateway/choreographer/pr_gate.py index bfd087cf..02752568 100644 --- a/roboco/services/gateway/choreographer/pr_gate.py +++ b/roboco/services/gateway/choreographer/pr_gate.py @@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any import structlog +from roboco.config import settings from roboco.foundation.policy import lifecycle as spec_module from roboco.foundation.policy import tracing as _tr from roboco.foundation.policy.batch import is_batch_root_subtask @@ -29,6 +30,10 @@ from roboco.foundation.policy.content import ( ) from roboco.services.gateway.choreographer import findings as findings_lib from roboco.services.gateway.choreographer.collision import build_collision_context +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) from roboco.services.gateway.envelope import Envelope from roboco.services.gateway.evidence_builder import render_findings from roboco.services.gateway.merge_chain import resolve_parent_branch @@ -1243,16 +1248,40 @@ class PRGateMixin(_Base): criteria + the task's OPEN findings (so they aren't crowded out by the full ledger's cap) + the full findings ledger (every status, newest round first) so the reviewer verifies prior rounds - item-by-item — parity with QA's ``_build_qa_claim_evidence``.""" + item-by-item — parity with QA's ``_build_qa_claim_evidence``. + + The diff + changed-files legs run bounded via ``run_bounded_leg`` + against ONE shared ``LegBudget`` for this build — a timeout skips + that piece and records a note in ``evidence_gaps`` instead of + hanging this advisory (non-gating) verb. + """ diff = "" files_changed: list[str] = [] + evidence_gaps: list[str] = [] if t.branch_name: gate_parent = await self._gate_diff_parent(t) - diff = await self.git.diff( - branch_name=t.branch_name, - preferred_parent=gate_parent, + budget = LegBudget(settings.evidence_assembly_timeout_seconds) + diff = await run_bounded_leg( + self.git.diff( + branch_name=t.branch_name, + preferred_parent=gate_parent, + ), + default="", + budget=budget, + leg="pr diff", + hint="review the PR diff on GitHub directly", + task_id=t.id, + gaps=evidence_gaps, + ) + files_changed = await run_bounded_leg( + self._gate_changed_files(t, gate_parent), + default=[], + budget=budget, + leg="files_changed", + hint="review the PR diff on GitHub directly", + task_id=t.id, + gaps=evidence_gaps, ) - files_changed = await self._gate_changed_files(t, gate_parent) open_findings = await findings_lib.open_findings_for_task( self.task.session, t.id ) @@ -1287,4 +1316,6 @@ class PRGateMixin(_Base): collision = await self._gate_collision_evidence(t, files_changed) if collision: evidence["collision_context"] = collision + if evidence_gaps: + evidence["evidence_gaps"] = evidence_gaps return evidence diff --git a/roboco/services/gateway/choreographer/qa.py b/roboco/services/gateway/choreographer/qa.py index 926a867b..4ac3ab67 100644 --- a/roboco/services/gateway/choreographer/qa.py +++ b/roboco/services/gateway/choreographer/qa.py @@ -52,6 +52,10 @@ from roboco.services.content_notes import apply_structured_note from roboco.services.gateway.choreographer import findings as findings_lib from roboco.services.gateway.choreographer._protocol import actor_context_fields from roboco.services.gateway.choreographer.collision import build_collision_context +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) from roboco.services.gateway.envelope import Envelope from roboco.services.gateway.evidence_builder import build_evidence_for_task @@ -190,19 +194,54 @@ class QAMixin(_Base): ).with_introspection(task=t, role=role_str) async def _qa_convention_findings( - self, qa_agent_id: UUID, t: Any + self, qa_agent_id: UUID, t: Any, *, timeout: float, gaps: list[str] ) -> list[dict[str, Any]]: """Convention-validator findings on the task's changed files (flag-gated). Empty when the subsystem is off; a validator that could not run surfaces a single explicit ``could_not_run`` entry rather than being dropped, so QA never mistakes a silent failure for a clean diff. + + Deliberately NOT wrapped in ``run_bounded_leg``. Nesting an outer + ``asyncio.wait_for`` around this call raced + ``conventions_check_for_task``'s own inner ``wait_for`` (which + starts LATER, after setup) — the outer cancellation could fire + first and land INSIDE the inner ``except TimeoutError:``'s + ``proc.kill()``/``proc.wait()`` cleanup, skipping it and leaking the + validator subprocess (reproduced live). ``conventions_check_for_task`` + already self-bounds the subprocess correctly via its own ``timeout`` + kwarg; this method just reads the result and, on a detected + timeout, records the gap itself instead of relying on an outer + wrapper to notice. + + ``timeout`` is this build's remaining share of the shared + evidence-assembly budget (``LegBudget``), already capped by the + caller at ``conventions_validator_advisory_timeout_seconds`` — so + this never exceeds its own designed ceiling but shrinks if earlier + legs (diff, list_changed_files) consumed most of the shared budget. + + The setup phase inside ``conventions_check_for_task`` (workspace + resolution + its own ``list_changed_files`` call) is not separately + wrapped here: every individual git op there is already bounded by + ``git_command_timeout_seconds`` inside ``_run_git`` (raising + ``GitTimeoutError`` on expiry, caught by ``conventions_check_for_task``'s + own resolution try/except), and by the time this runs the workspace + was already resolved (warmed) by the diff/list_changed_files legs + above in the same evidence build — so the one theoretically-unbounded + piece (a cold clone) never actually triggers here in practice. """ if not settings.conventions_enabled: return [] - result = await self.git.conventions_check_for_task(qa_agent_id, t) + result = await self.git.conventions_check_for_task( + qa_agent_id, t, timeout=timeout + ) if result.get("could_not_run"): reason = result.get("reason") or "validator could not run" + if "timed out" in reason: + gaps.append( + f"conventions findings unavailable: {reason} — review the " + "diff manually for architecture-convention issues" + ) return [{"could_not_run": True, "reason": reason}] return list(result.get("findings", [])) @@ -242,12 +281,38 @@ class QAMixin(_Base): ``add_files_modified`` HTTP path that populated files_modified is not called by the gateway ``commit()``, so the work_session list was always empty — QA saw no files even on real PRs. + + The slow legs (branch-fetch-backed diff, list_changed_files) run + bounded via ``run_bounded_leg`` against ONE shared ``LegBudget`` for + this whole build — a timeout skips that piece and records a note in + ``evidence_gaps`` instead of hanging this advisory (non-gating) verb + for the whole ``flow_verb_timeout_seconds`` budget. The conventions + leg self-bounds separately (see ``_qa_convention_findings``) but + still draws its ceiling from the same shared budget. """ files_changed: list[str] = [] diff_summary = "" + evidence_gaps: list[str] = [] + budget = LegBudget(settings.evidence_assembly_timeout_seconds) if t.branch_name: - diff_summary = await self.git.diff(branch_name=t.branch_name) - files_changed = await self.git.list_changed_files(branch_name=t.branch_name) + diff_summary = await run_bounded_leg( + self.git.diff(branch_name=t.branch_name), + default="", + budget=budget, + leg="pr diff", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, + ) + files_changed = await run_bounded_leg( + self.git.list_changed_files(branch_name=t.branch_name), + default=[], + budget=budget, + leg="files_changed", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, + ) journal_highlights = await self.evidence_repo.journal_highlights_for_task( task_id ) @@ -256,7 +321,15 @@ class QAMixin(_Base): # Leaf-only journals stay (include_ancestors defaults False above); # ancestor *descriptions* are the ask, not work-so-far. parent_context = await self.evidence_repo.ancestor_context_for_task(task_id) - convention_findings = await self._qa_convention_findings(qa_agent_id, t) + convention_findings = await self._qa_convention_findings( + qa_agent_id, + t, + timeout=min( + settings.conventions_validator_advisory_timeout_seconds, + budget.remaining(), + ), + gaps=evidence_gaps, + ) open_findings = await findings_lib.open_findings_for_task( self.task.session, t.id ) @@ -291,6 +364,7 @@ class QAMixin(_Base): parent_context=parent_context, collision_context=collision_context, video_context=self._qa_video_context(t), + evidence_gaps=evidence_gaps, ) async def _verify_qa_owner( diff --git a/roboco/services/gateway/content_actions.py b/roboco/services/gateway/content_actions.py index 7fb19ef8..7e1172e4 100644 --- a/roboco/services/gateway/content_actions.py +++ b/roboco/services/gateway/content_actions.py @@ -35,6 +35,10 @@ from roboco.foundation.policy.journaling import Scope as _Scope from roboco.models.base import TaskStatus from roboco.services.content_notes import content_type_for_role from roboco.services.gateway.choreographer import findings as findings_lib +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) from roboco.services.gateway.commit_validator import validate_commit_message from roboco.services.gateway.envelope import Envelope from roboco.services.gateway.evidence_builder import build_evidence_for_task @@ -5387,6 +5391,13 @@ class ContentActions: ``files_changed`` and ``pr_diff_summary`` are pulled from git (against the branch's parent — the authoritative source) rather than the latest commit's delta, so reviewers see the full multi-commit change set. + + The three slow legs (workspace branch fetch, diff, list_changed_files) + each run bounded via ``run_bounded_leg`` against ONE shared + ``LegBudget`` for this call — a timeout skips that piece and records + a note in ``evidence_gaps`` instead of hanging this advisory + (read-only, non-gating) verb for the whole ``flow_verb_timeout_seconds`` + budget. """ t = await self.task.get(task_id) if t is None: @@ -5416,18 +5427,53 @@ class ContentActions: await self.task.session.commit() except PendingRollbackError: await self.task.session.rollback() + evidence_gaps: list[str] = [] + budget = LegBudget(settings.evidence_assembly_timeout_seconds) if t.branch_name and t.work_session_id: - await self.workspace.fetch_branch_for_inspection( - agent_id=agent_id, branch_name=t.branch_name + # subprocess_timeout self-bounds the underlying git-fetch + # subprocess (on the shared DEFAULT asyncio executor, not + # git.py's dedicated pool) to roughly this leg's own share of + # the budget, so an abandoned wait_for doesn't leave the + # subprocess occupying a thread for up to workspace_clone_timeout + # (300s) after we've already given up on it. + await run_bounded_leg( + self.workspace.fetch_branch_for_inspection( + agent_id=agent_id, + branch_name=t.branch_name, + subprocess_timeout=budget.remaining(), + ), + default=None, + budget=budget, + leg="branch fetch", + hint=( + "the diff below may reflect a stale workspace; review " + "the PR diff on GitHub directly" + ), + task_id=task_id, + gaps=evidence_gaps, ) diff = "" files_changed: list[str] = [] if t.branch_name: - diff = await self.git.diff( - branch_name=t.branch_name, actor_agent_id=agent_id + diff = await run_bounded_leg( + self.git.diff(branch_name=t.branch_name, actor_agent_id=agent_id), + default="", + budget=budget, + leg="pr diff", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, ) - files_changed = await self.git.list_changed_files( - branch_name=t.branch_name, actor_agent_id=agent_id + files_changed = await run_bounded_leg( + self.git.list_changed_files( + branch_name=t.branch_name, actor_agent_id=agent_id + ), + default=[], + budget=budget, + leg="files_changed", + hint="review the PR diff on GitHub directly", + task_id=task_id, + gaps=evidence_gaps, ) journal_highlights = await self.evidence_repo.journal_highlights_for_task( task_id, include_ancestors=True @@ -5443,6 +5489,7 @@ class ContentActions: pr_diff_summary=diff, revision_findings=open_findings, parent_context=parent_context, + evidence_gaps=evidence_gaps, ) return Envelope.ok( status=str(t.status), diff --git a/roboco/services/gateway/evidence_builder.py b/roboco/services/gateway/evidence_builder.py index 132757eb..e54ad1cd 100644 --- a/roboco/services/gateway/evidence_builder.py +++ b/roboco/services/gateway/evidence_builder.py @@ -27,6 +27,7 @@ _EVIDENCE_OMIT_WHEN_EMPTY = ( "description", "collision_context", "video_context", + "evidence_gaps", ) @@ -69,6 +70,11 @@ class EvidencePayload: # the latest request_render preview + a verification instruction). None # for every non-video task; omitted from the dict when empty. video_context: dict[str, Any] | None = None + # Human-readable notes for an advisory-evidence leg (branch fetch, diff, + # conventions validator) that timed out and was skipped rather than + # hanging the verb — claim_review / claim_doc_task / claim_gate_review + # only. Empty (and omitted) on the normal path. + evidence_gaps: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: data = asdict(self) @@ -173,6 +179,7 @@ def build_evidence_for_task( parent_context: list[dict[str, Any]] | None = None, collision_context: list[dict[str, Any]] | None = None, video_context: dict[str, Any] | None = None, + evidence_gaps: list[str] | None = None, ) -> EvidencePayload: """Compose an EvidencePayload from a Task model + supplemental data. @@ -186,6 +193,9 @@ def build_evidence_for_task( verbatim so this module stays DB-free. ``video_context`` is the prebuilt video-artifact block (caller-assembled from the task's ``video_draft`` marker + render preview); passed through verbatim. + ``evidence_gaps`` carries the caller's bounded-leg timeout notes + (``choreographer.evidence_legs.run_bounded_leg``); empty on the normal + path. """ return EvidencePayload( pr_number=task.pr_number, @@ -203,6 +213,7 @@ def build_evidence_for_task( prior_findings=render_findings(prior_findings), collision_context=list(collision_context or []), video_context=video_context, + evidence_gaps=list(evidence_gaps or []), ) diff --git a/roboco/services/git.py b/roboco/services/git.py index 25013f36..d4f72529 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -6151,6 +6151,7 @@ class GitService(BaseService): task: Any, *, preferred_parent: str | None = None, + timeout: float | None = None, ) -> dict[str, Any]: """Run the conventions validator on a task's changed files. @@ -6166,6 +6167,13 @@ class GitService(BaseService): ``preferred_parent`` threads to ``list_changed_files`` — the in-path PR-review gate's cross-team parent (see ``diff``'s docstring). + ``timeout`` overrides the validator subprocess's default budget + (``_CONVENTIONS_VALIDATOR_TIMEOUT_SECONDS``) — the advisory + ``claim_review`` path passes ``settings. + conventions_validator_advisory_timeout_seconds`` here; every + fail-closed caller (``i_am_done``, ``pr_pass``) omits it and keeps + the longer hardcoded cap. + The changed-file LIST above comes from git objects (``list_changed_files`` fetches + diffs ``origin/``); the validator below reads CONTENT off the physical worktree, which only ``_ensure_worktree_for_commit`` @@ -6214,11 +6222,16 @@ class GitService(BaseService): # content and false-passes on newly-added files. workspace = self._worktree_for_task(clone_root, require_uuid(task.id)) await self._ensure_worktree_for_commit(clone_root, workspace, branch) - return await self._run_conventions_validator(workspace, changed) + return await self._run_conventions_validator( + workspace, changed, timeout=timeout + ) async def _run_conventions_validator( - self, workspace: Path, files: list[str] + self, workspace: Path, files: list[str], *, timeout: float | None = None ) -> dict[str, Any]: + effective_timeout = ( + timeout if timeout is not None else _CONVENTIONS_VALIDATOR_TIMEOUT_SECONDS + ) proc = await asyncio.create_subprocess_exec( sys.executable, "-m", @@ -6234,7 +6247,7 @@ class GitService(BaseService): try: out, err = await asyncio.wait_for( proc.communicate(), - timeout=_CONVENTIONS_VALIDATOR_TIMEOUT_SECONDS, + timeout=effective_timeout, ) except TimeoutError: # Fail closed (could_not_run=True → block gate refuses the submit), @@ -6245,10 +6258,7 @@ class GitService(BaseService): return { "findings": [], "could_not_run": True, - "reason": ( - f"validator timed out after " - f"{_CONVENTIONS_VALIDATOR_TIMEOUT_SECONDS}s" - ), + "reason": (f"validator timed out after {effective_timeout}s"), } if proc.returncode != 0: reason = err.decode(errors="replace").strip() or "validator crashed" diff --git a/roboco/services/prompter.py b/roboco/services/prompter.py index ed22a4b9..ec2d26b1 100644 --- a/roboco/services/prompter.py +++ b/roboco/services/prompter.py @@ -509,7 +509,7 @@ class PrompterService: if not raw: return () deps: list[int] = [] - for item in raw if isinstance(raw, (list, tuple)) else [raw]: + for item in raw if isinstance(raw, list | tuple) else [raw]: try: deps.append(int(str(item).strip())) except (TypeError, ValueError) as exc: diff --git a/roboco/services/workspace.py b/roboco/services/workspace.py index 6f006e5f..6cd51c81 100644 --- a/roboco/services/workspace.py +++ b/roboco/services/workspace.py @@ -40,6 +40,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from roboco.config import settings from roboco.db.tables import AgentTable +from roboco.exceptions import GitTimeoutError from roboco.logging import get_logger from roboco.models.base import Team from roboco.models.env_branches import head_branch @@ -2260,6 +2261,7 @@ class WorkspaceService: *, agent_id: UUID, branch_name: str, + subprocess_timeout: float | None = None, ) -> Path: """Fetch `branch_name` into the inspecting agent's workspace. @@ -2268,10 +2270,22 @@ class WorkspaceService: 1. Resolves the project from the branch (via the owning task). 2. Ensures a healthy workspace for `agent_id` on that project - (clones if missing — same path as the agent's first claim). + (clones if missing — same path as the agent's first claim). This + CLONE-CREATION step always keeps its own `workspace_clone_timeout` + (300s) budget regardless of `subprocess_timeout` below — a fresh + clone genuinely needs that much room. 3. Runs `git fetch origin ` with the project token so the branch ref is locally available for `git diff`. + `subprocess_timeout` overrides ONLY the fetch subprocess in step 3 + (default `settings.workspace_clone_timeout`, 300s, for every + existing caller that omits it). The advisory `evidence()` call site + passes its own leg budget so a hung fetch self-terminates near that + window instead of occupying a thread on the shared DEFAULT asyncio + executor (`asyncio.to_thread` here, NOT git.py's dedicated + `_GIT_EXECUTOR`) for up to 300s after the caller already gave up + waiting on it. + Returns the workspace path so the caller can chain checkout/diff operations if needed. """ @@ -2306,17 +2320,31 @@ class WorkspaceService: basic = base64.b64encode(f"x-access-token:{git_token}".encode()).decode() prefix = ["-c", f"http.extraheader=Authorization: Basic {basic}"] + effective_timeout = ( + subprocess_timeout + if subprocess_timeout is not None + else settings.workspace_clone_timeout + ) + def _do_fetch() -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", *prefix, "fetch", "origin", branch_name], cwd=str(workspace), capture_output=True, text=True, - timeout=settings.workspace_clone_timeout, + timeout=effective_timeout, check=False, ) - result = await asyncio.to_thread(_do_fetch) + try: + result = await asyncio.to_thread(_do_fetch) + except subprocess.TimeoutExpired as exc: + # Mirrors _run_git's own TimeoutExpired -> GitTimeoutError + # conversion (git.py) so a bounded caller (run_bounded_leg) + # catches this the same way as every other git-touching leg. + raise GitTimeoutError( + f"fetch origin {branch_name}", int(effective_timeout) + ) from exc if result.returncode != 0: logger.warning( "fetch_branch_for_inspection: fetch returned non-zero", diff --git a/tests/foundation/test_lifecycle_consumer_parity.py b/tests/foundation/test_lifecycle_consumer_parity.py index 86011d05..4baa01ae 100644 --- a/tests/foundation/test_lifecycle_consumer_parity.py +++ b/tests/foundation/test_lifecycle_consumer_parity.py @@ -989,6 +989,19 @@ async def test_escalate_up_matches_spec(role: str, status: str) -> None: id=agent_id, role=role, team="backend", slug=None, escalation_target="main-pm" ) task_svc.escalate.return_value = after + # escalate_up's own _ensure_pm_decision write-then-gate opens a + # session.begin_nested() savepoint unconditionally (even on the + # fresh-decision no-write path below) — an unshaped AsyncMock's + # auto-attribute return doesn't support `async with`, which orphans + # the mock's internal coroutine (AsyncMockMixin._execute_mock_call + # never awaited). + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) journal_svc = AsyncMock() # journal:decision is not on the spec — satisfy it so the verb-specific # preflight does not surface a non-spec tracing_gap on the allowed branch. diff --git a/tests/integration/test_a2a_routes.py b/tests/integration/test_a2a_routes.py index fcecff33..f1196e53 100644 --- a/tests/integration/test_a2a_routes.py +++ b/tests/integration/test_a2a_routes.py @@ -325,7 +325,6 @@ async def test_send_message_with_task_id_response(a2a_route_client: dict) -> Non async def test_send_message_response_invalid_task_id( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -352,7 +351,6 @@ async def test_send_message_response_invalid_task_id( async def test_send_message_response_task_not_found( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -379,7 +377,6 @@ async def test_send_message_response_task_not_found( async def test_send_message_create_notification_success( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -403,7 +400,6 @@ async def test_send_message_create_notification_success( @pytest.mark.asyncio async def test_send_message_permission_error(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -427,7 +423,6 @@ async def test_send_message_permission_error(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_send_message_value_error(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -454,7 +449,6 @@ async def test_send_message_value_error(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_cancel_task_success(a2a_route_client: dict) -> None: - a2a_task = A2ATask.model_validate( { "id": str(a2a_route_client["task"].id), @@ -484,7 +478,6 @@ async def test_cancel_task_success(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: @@ -502,7 +495,6 @@ async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_cancel_task_not_found(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: @@ -638,7 +630,6 @@ async def test_send_message_uses_authenticated_identity_not_client_from_agent( async def test_chat_create_conversation_access_denied( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -667,7 +658,6 @@ async def test_chat_create_conversation_access_denied( async def test_chat_create_conversation_success( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] conv_id = uuid4() with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: @@ -707,7 +697,6 @@ async def test_chat_create_conversation_success( async def test_chat_create_conversation_refresh_failed( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -744,7 +733,6 @@ async def test_chat_create_conversation_refresh_failed( @pytest.mark.asyncio async def test_get_conversation_not_found(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -758,7 +746,6 @@ async def test_get_conversation_not_found(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_get_conversation_success(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] conv_id = uuid4() with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: @@ -790,7 +777,6 @@ async def test_get_conversation_success(a2a_route_client: dict) -> None: async def test_close_conversation_value_error( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -806,7 +792,6 @@ async def test_close_conversation_value_error( @pytest.mark.asyncio async def test_close_conversation_success(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -822,7 +807,6 @@ async def test_close_conversation_success(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_list_chat_messages(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] msg = SimpleNamespace( id=uuid4(), @@ -853,7 +837,6 @@ async def test_list_chat_messages(a2a_route_client: dict) -> None: async def test_send_chat_message_value_error( a2a_route_client: dict, ) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -869,7 +852,6 @@ async def test_send_chat_message_value_error( @pytest.mark.asyncio async def test_send_chat_message_success(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] msg = SimpleNamespace( id=uuid4(), @@ -926,7 +908,6 @@ async def test_send_chat_message_over_budget_returns_403( @pytest.mark.asyncio async def test_mark_read(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -941,7 +922,6 @@ async def test_mark_read(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_get_task_conversations(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() @@ -956,7 +936,6 @@ async def test_get_task_conversations(a2a_route_client: dict) -> None: @pytest.mark.asyncio async def test_chat_list_with_status_filter(a2a_route_client: dict) -> None: - client = a2a_route_client["client"] with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() diff --git a/tests/integration/test_foundation_phase2_smoke.py b/tests/integration/test_foundation_phase2_smoke.py index 4d4efb17..29333a1c 100644 --- a/tests/integration/test_foundation_phase2_smoke.py +++ b/tests/integration/test_foundation_phase2_smoke.py @@ -18,7 +18,7 @@ def _enclosing_function(tree: ast.AST, lineno: int) -> str | None: candidate: str | None = None candidate_start = -1 for node in ast.walk(tree): - if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)): + if isinstance(node, ast.AsyncFunctionDef | ast.FunctionDef): start = node.lineno end = getattr(node, "end_lineno", None) or start if start <= lineno <= end and start > candidate_start: diff --git a/tests/integration/test_task_service_lifecycle_misc.py b/tests/integration/test_task_service_lifecycle_misc.py index 2613b723..d0de0010 100644 --- a/tests/integration/test_task_service_lifecycle_misc.py +++ b/tests/integration/test_task_service_lifecycle_misc.py @@ -884,7 +884,6 @@ async def test_soft_block_task_for_agent_full_flow( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: - svc = task_setup["svc"] task = await svc.create(_req(task_setup)) task.assigned_to = task_setup["agent_id"] @@ -929,7 +928,6 @@ async def test_soft_block_task_for_agent_full_flow( async def test_docs_complete_for_task_invokes_notification( task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: - svc = task_setup["svc"] doc = AgentTable( id=uuid4(), @@ -988,7 +986,6 @@ async def test_docs_complete_for_task_invokes_notification( async def test_escalate_to_ceo_for_agent_invokes_notification( task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: - svc = task_setup["svc"] pm = AgentTable( id=uuid4(), @@ -1054,7 +1051,6 @@ async def test_escalate_to_ceo_for_agent_invokes_notification( async def test_claim_task_for_agent_commits_and_returns( task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: - svc = task_setup["svc"] task = await svc.create(_req(task_setup)) task.branch_name = "feature/backend/x" @@ -1088,7 +1084,6 @@ async def test_claim_task_for_agent_commits_and_returns( async def test_complete_task_for_agent_commits( task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: - svc = task_setup["svc"] pm = AgentTable( id=uuid4(), @@ -1138,7 +1133,6 @@ async def test_complete_task_for_agent_commits( async def test_substitute_task_for_agent_runs_update( task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: - svc = task_setup["svc"] task = await svc.create(_req(task_setup)) task.assigned_to = task_setup["agent_id"] diff --git a/tests/integration/test_tasks_routes.py b/tests/integration/test_tasks_routes.py index 801dcc8b..9419e49f 100644 --- a/tests/integration/test_tasks_routes.py +++ b/tests/integration/test_tasks_routes.py @@ -1182,7 +1182,6 @@ async def test_unblock_task_not_blocked_returns_400(task_client: dict) -> None: @pytest.mark.asyncio async def test_unblock_task_forbidden(task_client: dict) -> None: - other = await _seed_agent(task_client) task = _seed_task(task_client, status=TaskStatus.BLOCKED, assigned_to=other.id) await task_client["db"].flush() @@ -1615,7 +1614,6 @@ async def test_activate_unknown_returns_4xx(task_client: dict) -> None: @pytest.mark.asyncio async def test_activate_developer_forbidden(task_client: dict) -> None: - app = task_client["client"]._transport.app async def _override_agent() -> AgentContext: @@ -1668,7 +1666,6 @@ async def test_get_team_tasks_unauthorized(task_client: dict) -> None: async def test_get_task_stats_by_team_developer_forbidden( task_client: dict, ) -> None: - app = task_client["client"]._transport.app async def _override_agent() -> AgentContext: @@ -1779,7 +1776,6 @@ async def test_delete_task_forbidden_non_creator(task_client: dict) -> None: @pytest.mark.asyncio async def test_cancel_developer_forbidden(task_client: dict) -> None: - task = _seed_task(task_client) await task_client["db"].flush() app = task_client["client"]._transport.app diff --git a/tests/unit/api/test_websocket_handler_cleanup.py b/tests/unit/api/test_websocket_handler_cleanup.py index 7abcef34..fc52259a 100644 --- a/tests/unit/api/test_websocket_handler_cleanup.py +++ b/tests/unit/api/test_websocket_handler_cleanup.py @@ -84,7 +84,6 @@ async def test_system_stream_disconnects_on_cancelled_error() -> None: async def test_notification_stream_disconnects_on_non_disconnect_exception( monkeypatch: pytest.MonkeyPatch, ) -> None: - agent_id = uuid4() mgr = ConnectionManager() ws = _mock_ws_for_receive(RuntimeError("transport reset")) @@ -109,7 +108,6 @@ async def test_notification_stream_disconnects_on_non_disconnect_exception( async def test_agent_stream_disconnects_on_non_disconnect_exception( monkeypatch: pytest.MonkeyPatch, ) -> None: - target_id = uuid4() viewer_id = uuid4() mgr = ConnectionManager() diff --git a/tests/unit/api/test_websocket_idle_timeout.py b/tests/unit/api/test_websocket_idle_timeout.py index 813b23cb..76610240 100644 --- a/tests/unit/api/test_websocket_idle_timeout.py +++ b/tests/unit/api/test_websocket_idle_timeout.py @@ -89,7 +89,6 @@ async def test_system_stream_reaps_silent_socket_after_idle_timeout() -> None: async def test_notification_stream_reaps_silent_socket_after_idle_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: - agent_id = uuid4() mgr = ConnectionManager() hang_future: asyncio.Future[str] = asyncio.Future() @@ -109,7 +108,6 @@ async def test_notification_stream_reaps_silent_socket_after_idle_timeout( async def test_agent_stream_reaps_silent_socket_after_idle_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: - target_id = uuid4() viewer_id = uuid4() mgr = ConnectionManager() diff --git a/tests/unit/gateway/test_choreographer_pm.py b/tests/unit/gateway/test_choreographer_pm.py index ef7740ba..fbf2e5bb 100644 --- a/tests/unit/gateway/test_choreographer_pm.py +++ b/tests/unit/gateway/test_choreographer_pm.py @@ -978,18 +978,27 @@ async def test_escalate_up_survives_journal_write_lock_timeout() -> None: with no rollback/savepoint, poisoning the session so the very next attribute touch (``_escalate_up_preflight`` reading ``t.id``) raised an unhandled ``PendingRollbackError``. The write is now savepoint-guarded - (``begin_nested()``): the failure is contained, the verb falls through - cleanly to the normal tracing_gap rejection (no decision was actually - persisted), and the task stays fully readable — no unhandled exception - escapes ``escalate_up``.""" + (``begin_nested()``): the failure is contained and no unhandled + exception escapes ``escalate_up``. + + Round-2 fix (transient-failure gate bypass): a lock-timeout with a + non-empty rationale already in hand (``reason``) no longer falls through + to a tracing_gap rejection either — the PM answered the gate's actual + question (why); the write was only ever a convenience. Laundering + transient DB congestion into a rejected/blocked PM verb is exactly the + bug this closes — see test_pm_decision_transient_failure_* below for the + gate-helper-level unit coverage. + """ pm_id = uuid4() task_id = uuid4() t = MagicMock(id=task_id, status="blocked", assigned_to=pm_id, team="backend") + after = MagicMock(**{**t.__dict__, "assigned_to": uuid4()}) task_svc = AsyncMock() task_svc.get.return_value = t task_svc.agent_for.return_value = MagicMock( role="cell_pm", escalation_target="main-pm" ) + task_svc.escalate.return_value = after journal_svc = AsyncMock() journal_svc.has_decision_for_task.return_value = False journal_svc.latest_decision_at.return_value = None @@ -1006,11 +1015,11 @@ async def test_escalate_up_survives_journal_write_lock_timeout() -> None: # The savepoint was actually engaged — proves the fix is wired in, not # merely that AsyncMock happened to swallow the raise on its own. task_svc.session.begin_nested.assert_called() - # No unhandled exception escaped escalate_up: the gate falls through to - # its normal clean rejection since the decision write never landed. + # No unhandled exception escaped escalate_up, AND the gate is satisfied + # by the verb's own rationale instead of rejecting a DB hiccup. body = env.as_dict() - assert body["error"] == "tracing_gap" - assert "journal:decision" in body["missing"] + assert body["error"] is None, body + task_svc.escalate.assert_awaited_once() # The task is still fully readable afterward — this is exactly where # the production trace crashed with PendingRollbackError on t.id. assert t.id == task_id diff --git a/tests/unit/gateway/test_conventions_gate_i_am_done.py b/tests/unit/gateway/test_conventions_gate_i_am_done.py index 1ef7fba6..7e33c86c 100644 --- a/tests/unit/gateway/test_conventions_gate_i_am_done.py +++ b/tests/unit/gateway/test_conventions_gate_i_am_done.py @@ -109,3 +109,18 @@ async def test_gate_records_findings_even_when_blocking( env = await c._conventions_gate(_ctx()) assert env is not None # still blocks assert recorded and recorded[0] is _BLOCK_RESULT + + +@pytest.mark.asyncio +async def test_gate_never_overrides_the_fail_closed_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """i_am_done's conventions gate is fail-closed and must keep the + validator's hardcoded 120s cap — unlike claim_review's advisory path, it + must never pass a ``timeout`` override down to ``conventions_check_for_task``.""" + monkeypatch.setattr(settings, "conventions_enabled", True) + c = _make_choreographer(check_result={"findings": [], "could_not_run": False}) + await c._conventions_gate(_ctx()) + check = c.git.conventions_check_for_task + check.assert_awaited_once() + assert "timeout" not in check.await_args.kwargs diff --git a/tests/unit/gateway/test_conventions_gate_pr_pass.py b/tests/unit/gateway/test_conventions_gate_pr_pass.py index 327b121a..d6f1c35d 100644 --- a/tests/unit/gateway/test_conventions_gate_pr_pass.py +++ b/tests/unit/gateway/test_conventions_gate_pr_pass.py @@ -125,3 +125,18 @@ async def test_pr_pass_guard_inert_when_flag_off( monkeypatch.setattr(settings, "conventions_enabled", False) c = _make_choreographer(check_result=_BLOCK_RESULT) assert await c._conventions_guard(uuid4(), MagicMock(), {}) is None + + +@pytest.mark.asyncio +async def test_pr_pass_guard_never_overrides_the_fail_closed_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The pr_pass gate is fail-closed and must keep the validator's + hardcoded 120s cap — unlike claim_review's advisory path, it must + never pass a ``timeout`` override down to ``conventions_check_for_task``.""" + monkeypatch.setattr(settings, "conventions_enabled", True) + c = _make_choreographer(check_result={"findings": [], "could_not_run": False}) + await c._conventions_guard(uuid4(), MagicMock(), {}) + check = c.git.conventions_check_for_task + check.assert_awaited_once() + assert "timeout" not in check.await_args.kwargs diff --git a/tests/unit/gateway/test_conventions_in_qa_evidence.py b/tests/unit/gateway/test_conventions_in_qa_evidence.py index 3b37fe20..9c3b56ce 100644 --- a/tests/unit/gateway/test_conventions_in_qa_evidence.py +++ b/tests/unit/gateway/test_conventions_in_qa_evidence.py @@ -31,7 +31,12 @@ async def test_findings_surfaced_when_flag_on(monkeypatch: pytest.MonkeyPatch) - monkeypatch.setattr(settings, "conventions_enabled", True) findings = [{"file": "x.py", "line": 1, "level": "warn", "fix_hint": "h"}] c = _make_choreographer(check_result={"findings": findings, "could_not_run": False}) - assert await c._qa_convention_findings(uuid4(), MagicMock()) == findings + gaps: list[str] = [] + assert ( + await c._qa_convention_findings(uuid4(), MagicMock(), timeout=30.0, gaps=gaps) + == findings + ) + assert gaps == [] @pytest.mark.asyncio @@ -40,21 +45,32 @@ async def test_empty_when_flag_off(monkeypatch: pytest.MonkeyPatch) -> None: c = _make_choreographer( check_result={"findings": [{"file": "x"}], "could_not_run": False} ) - assert await c._qa_convention_findings(uuid4(), MagicMock()) == [] + gaps: list[str] = [] + assert ( + await c._qa_convention_findings(uuid4(), MagicMock(), timeout=30.0, gaps=gaps) + == [] + ) + assert gaps == [] @pytest.mark.asyncio async def test_could_not_run_surfaced_as_single_entry( monkeypatch: pytest.MonkeyPatch, ) -> None: + """A non-timeout could_not_run reason ("boom") stays fail-open in + convention_findings but must NOT also spam evidence_gaps — only a + detected timeout does (see test_claim_review_conventions_timeout_ + degrades_with_gap in test_evidence_assembly_bounded_legs.py).""" monkeypatch.setattr(settings, "conventions_enabled", True) c = _make_choreographer( check_result={"findings": [], "could_not_run": True, "reason": "boom"} ) - out = await c._qa_convention_findings(uuid4(), MagicMock()) + gaps: list[str] = [] + out = await c._qa_convention_findings(uuid4(), MagicMock(), timeout=30.0, gaps=gaps) assert len(out) == 1 assert out[0]["could_not_run"] is True assert out[0]["reason"] == "boom" + assert gaps == [] def _stub_task() -> MagicMock: diff --git a/tests/unit/gateway/test_evidence_assembly_bounded_legs.py b/tests/unit/gateway/test_evidence_assembly_bounded_legs.py new file mode 100644 index 00000000..8362ca4d --- /dev/null +++ b/tests/unit/gateway/test_evidence_assembly_bounded_legs.py @@ -0,0 +1,763 @@ +"""Bounded advisory-evidence legs on claim_review / claim_doc_task / +claim_gate_review / evidence() / i_am_done's success envelope. + +Live bug: claim-evidence assembly's slow legs (branch-fetch-backed diff, +list_changed_files, the conventions-validator subprocess) had no per-leg +budget, so a hung leg silently ate the whole ``flow_verb_timeout_seconds`` +(120s) and died as a FlowVerbTimeout 504 — holding every row the request +touched for the duration. Fix: each slow leg runs bounded via +``run_bounded_leg`` (``asyncio.wait_for``) against a SHARED ``LegBudget`` per +build; a timeout skips that piece, records a human-readable note in the +evidence's ``evidence_gaps``, and lets the claim verb succeed with partial +evidence instead of hanging. + +Adversarial-review follow-up (round 2) covers four confirmed gaps: +1. ``run_bounded_leg`` must catch ``GitTimeoutError`` too (``_run_git``'s own + internal subprocess bound — NOT a ``TimeoutError`` subclass, and usually + the FIRST bound to trip since it defaults to 30s, shorter than a leg's + own budget) — every timeout-shaped test below is parametrized over both + exception shapes. +2. The conventions leg no longer wraps ``_qa_convention_findings`` in an + outer ``run_bounded_leg`` — that raced ``conventions_check_for_task``'s + own inner timeout+cleanup and leaked the validator subprocess. It now + self-bounds via the ``timeout`` kwarg alone and reports its own gap. +3. ``fetch_branch_for_inspection`` takes an optional ``subprocess_timeout`` + so its fetch subprocess self-bounds near the leg's own budget instead of + occupying a thread on the shared default executor for up to 300s. +4. A shared ``LegBudget`` (one per evidence build) makes every leg's + ``wait_for`` draw from one TOTAL budget instead of getting its own full + allotment — summed per-leg budgets can no longer exceed the total. +""" + +from __future__ import annotations + +import asyncio +import time +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.config import settings +from roboco.exceptions import GitCommandError, GitTimeoutError +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps +from roboco.services.gateway.choreographer.evidence_legs import ( + LegBudget, + run_bounded_leg, +) + +# Every timeout-shaped test is parametrized over both real timeout shapes: +# asyncio's own cancellation-converted TimeoutError, and GitTimeoutError +# (GitService._run_git's own internal subprocess bound — a GitError/ +# RobocoError subclass, NOT a TimeoutError subclass, and the most common +# real-world single-hung-git-call shape since it defaults to a SHORTER +# window, 30s, than a leg's own budget). +_TIMEOUT_EXCEPTIONS = ( + TimeoutError("hung"), + GitTimeoutError("git diff", 30), +) +_TIMEOUT_IDS = ("asyncio_timeout", "git_timeout") + + +# --------------------------------------------------------------------------- +# run_bounded_leg / LegBudget themselves +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_run_bounded_leg_passes_through_on_success() -> None: + async def fast() -> str: + return "value" + + gaps: list[str] = [] + result = await run_bounded_leg( + fast(), + default="fallback", + budget=LegBudget(5.0), + leg="unit leg", + hint="check manually", + task_id=uuid4(), + gaps=gaps, + ) + assert result == "value" + assert gaps == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_run_bounded_leg_degrades_to_default_on_timeout(exc: Exception) -> None: + async def hangs() -> str: + raise exc + + gaps: list[str] = [] + result = await run_bounded_leg( + hangs(), + default="fallback", + budget=LegBudget(5.0), + leg="unit leg", + hint="check manually", + task_id=uuid4(), + gaps=gaps, + ) + assert result == "fallback" + assert len(gaps) == 1 + assert "unit leg unavailable" in gaps[0] + assert "check manually" in gaps[0] + + +@pytest.mark.asyncio +async def test_run_bounded_leg_actually_bounds_a_slow_coroutine() -> None: + """A genuinely slow (not pre-raised) coroutine is cancelled at the + budget's deadline, not awaited to completion.""" + + async def slow() -> str: + await asyncio.sleep(10) + return "too late" + + gaps: list[str] = [] + result = await run_bounded_leg( + slow(), + default="fallback", + budget=LegBudget(0.05), + leg="unit leg", + hint="check manually", + task_id=uuid4(), + gaps=gaps, + ) + assert result == "fallback" + assert len(gaps) == 1 + + +@pytest.mark.asyncio +async def test_run_bounded_leg_other_git_error_still_propagates() -> None: + """A real command failure (not a timeout) is NOT a degrade case — it + must still propagate uncaught, same as any other unexpected exception.""" + + async def fails() -> str: + raise GitCommandError("git diff", "fatal: bad revision") + + gaps: list[str] = [] + with pytest.raises(GitCommandError): + await run_bounded_leg( + fails(), + default="fallback", + budget=LegBudget(5.0), + leg="unit leg", + hint="check manually", + task_id=uuid4(), + gaps=gaps, + ) + assert gaps == [] + + +def test_leg_budget_remaining_shrinks_over_time() -> None: + # Total well above _MIN_LEG_SECONDS (1.0) so the floor never engages + # here — otherwise both readings would clamp to 1.0 and look equal. + budget = LegBudget(3.0) + first = budget.remaining() + time.sleep(0.1) + second = budget.remaining() + assert second < first + assert first == pytest.approx(3.0, abs=0.05) + assert (first - second) == pytest.approx(0.1, abs=0.05) + + +def test_leg_budget_floors_at_minimum() -> None: + """A budget already past its deadline still yields a positive window + (the floor) rather than 0 or a negative timeout — a leg always gets a + real chance to run, even a badly-exhausted one.""" + budget = LegBudget(0.01) + time.sleep(0.05) + assert budget.remaining() == pytest.approx(1.0, abs=0.05) + + +@pytest.mark.asyncio +async def test_run_bounded_leg_shares_one_shrinking_budget_across_legs() -> None: + """Three legs sharing ONE LegBudget: the first two finish fast and + consume real budget; the last two never finish on their own and get + progressively SMALLER windows (shrinking, not each getting the full + total) — both record their own gap, and total wall time stays bounded + near the shared total instead of the naive per-leg sum (0.1+0.1+5+5s). + """ + budget = LegBudget(1.2) + gaps: list[str] = [] + + async def _takes(seconds: float) -> str: + await asyncio.sleep(seconds) + return "done" + + start = time.monotonic() + remaining_before_1 = budget.remaining() + r1 = await run_bounded_leg( + _takes(0.1), + default="gap1", + budget=budget, + leg="leg1", + hint="h", + task_id="t", + gaps=gaps, + ) + remaining_before_2 = budget.remaining() + r2 = await run_bounded_leg( + _takes(5.0), + default="gap2", + budget=budget, + leg="leg2", + hint="h", + task_id="t", + gaps=gaps, + ) + remaining_before_3 = budget.remaining() + r3 = await run_bounded_leg( + _takes(5.0), + default="gap3", + budget=budget, + leg="leg3", + hint="h", + task_id="t", + gaps=gaps, + ) + elapsed = time.monotonic() - start + expected_gap_count = 2 # leg2 + leg3 both timed out; leg1 completed + + assert r1 == "done" + assert r2 == "gap2" + assert r3 == "gap3" + assert len(gaps) == expected_gap_count + assert "leg2" in gaps[0] + assert "leg3" in gaps[1] + # Each leg's own remaining() reading is strictly smaller than the last + # — the shared deadline never resets. + assert remaining_before_1 > remaining_before_2 > remaining_before_3 + # ponytail: the floor (max 1.0s) can inflate the LAST leg's window past + # what a naive "budget minus elapsed" would give once the deadline is + # already exhausted — a single floor engagement caps the worst-case + # overage at _MIN_LEG_SECONDS (1.0s), so budget total + ~1.3s is a safe, + # honest ceiling rather than a strict `<= budget` bound. Upgrade path: + # make the floor configurable if a caller ever needs a tighter cap. + budget_total_seconds = 1.2 + floor_overage_tolerance_seconds = 1.3 + assert elapsed <= budget_total_seconds + floor_overage_tolerance_seconds + # And it's nowhere near the naive per-leg-gets-its-own-full-timeout sum + # (0.1 + 5.0 + 5.0 = 10.1s) that pre-LegBudget behavior would produce. + naive_per_leg_sum_seconds = 5.0 + assert elapsed < naive_per_leg_sum_seconds + + +# --------------------------------------------------------------------------- +# Shared choreographer test harness (mirrors test_choreographer_qa.py / +# test_claim_doc_task_checkout.py / test_claim_gate_review_guards.py) +# --------------------------------------------------------------------------- + + +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) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + _ldef = base["journal"].latest_decision_at.return_value + if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): + base["journal"].latest_decision_at.return_value = datetime.now(UTC) + return ChoreographerDeps(**base) + + +def _stub_empty_ledger(session: MagicMock) -> None: + session.execute = AsyncMock( + return_value=MagicMock( + scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))) + ) + ) + + +# --------------------------------------------------------------------------- +# claim_review (qa.py) +# --------------------------------------------------------------------------- + +_PR_NUMBER = 8 +_PR_URL = "https://github.com/x/y/pull/8" + + +def _qa_task(task_id: Any) -> MagicMock: + return MagicMock( + id=task_id, + status="awaiting_qa", + assigned_to=None, + pr_number=_PR_NUMBER, + pr_url=_PR_URL, + commits=[{"sha": "abc123", "message": "feat: x"}], + team="backend", + branch_name="feature/backend/abc--def", + work_session_id=uuid4(), + documents=[], + dev_notes="implemented x", + acceptance_criteria=["AC1"], + acceptance_criteria_status=[ + {"criterion": "AC1", "referencing_artifact_id": "abc123"}, + ], + parent_task_id=None, + ) + + +def _qa_harness(git_svc: AsyncMock) -> tuple[Choreographer, Any, Any]: + """Does NOT touch ``settings.conventions_enabled`` — callers that care + set it themselves via their own ``monkeypatch`` fixture; a shared + forced-False here would silently clobber a caller's forced-True set + moments earlier (``monkeypatch.setattr`` doesn't stack, last write + wins), which is exactly what broke the conventions-specific tests.""" + qa_id = uuid4() + task_id = uuid4() + t_initial = _qa_task(task_id) + t_claimed = MagicMock(**{**t_initial.__dict__, "assigned_to": qa_id}) + + task_svc = AsyncMock() + task_svc.get.return_value = t_initial + task_svc.agent_for.return_value = MagicMock(role="qa", team="backend") + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.qa_claim.return_value = t_claimed + _stub_empty_ledger(task_svc.session) + + deps = _make_deps(task=task_svc, git=git_svc) + return Choreographer(deps), qa_id, task_id + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_claim_review_diff_timeout_degrades_with_gap( + monkeypatch: pytest.MonkeyPatch, exc: Exception +) -> None: + """A hung git.diff on claim_review must not hang the verb: it degrades + to an empty diff, records the gap, and the OTHER leg (list_changed_files) + still comes through untouched.""" + monkeypatch.setattr(settings, "conventions_enabled", False) + git_svc = AsyncMock() + git_svc.diff.side_effect = exc + git_svc.list_changed_files.return_value = ["README.md"] + c, qa_id, task_id = _qa_harness(git_svc) + + env = await c.claim_review(qa_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" in ev + assert len(ev["evidence_gaps"]) == 1 + assert "pr diff unavailable" in ev["evidence_gaps"][0] + + +@pytest.mark.asyncio +async def test_claim_review_conventions_timeout_degrades_with_gap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """conventions_check_for_task's OWN internal timeout (proc.kill()'d and + reaped inside git.py, never raising) surfaces as could_not_run=True with + a "timed out" reason — not an exception. The advisory call site + (_qa_convention_findings) detects that shape and records the gap + itself; NO outer run_bounded_leg wraps this leg (that's the fix — see + module docstring point 2).""" + monkeypatch.setattr(settings, "conventions_enabled", True) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + git_svc.conventions_check_for_task.return_value = { + "findings": [], + "could_not_run": True, + "reason": "validator timed out after 30.0s", + } + c, qa_id, task_id = _qa_harness(git_svc) + + env = await c.claim_review(qa_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert ev["convention_findings"] == [ + {"could_not_run": True, "reason": "validator timed out after 30.0s"} + ] + assert "evidence_gaps" in ev + assert any("conventions findings unavailable" in g for g in ev["evidence_gaps"]) + # The advisory (shorter) ceiling reached the validator call, not the + # fail-closed i_am_done/pr_pass default (None -> hardcoded 120s). + git_svc.conventions_check_for_task.assert_awaited_once() + call_kwargs = git_svc.conventions_check_for_task.await_args.kwargs + assert ( + call_kwargs["timeout"] + <= settings.conventions_validator_advisory_timeout_seconds + ) + assert call_kwargs["timeout"] > 0 + + +@pytest.mark.asyncio +async def test_claim_review_conventions_non_timeout_could_not_run_no_gap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A genuine resolution failure (not a timeout) still surfaces in + convention_findings (existing fail-open shape) but must NOT also spam + evidence_gaps — that's reserved for actual degraded-advisory-leg notes.""" + monkeypatch.setattr(settings, "conventions_enabled", True) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + git_svc.conventions_check_for_task.return_value = { + "findings": [], + "could_not_run": True, + "reason": "resolution failed: NotFoundError: Branch not found", + } + c, qa_id, task_id = _qa_harness(git_svc) + + env = await c.claim_review(qa_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["convention_findings"][0]["could_not_run"] is True + assert "evidence_gaps" not in ev + + +@pytest.mark.asyncio +async def test_qa_convention_findings_not_cancelled_by_outer_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression for the orphaned-subprocess bug: even with a tiny shared + evidence-assembly budget, _qa_convention_findings must NOT be cut short + by an outer wait_for — it awaits conventions_check_for_task to + completion. A slow-but-real mock (0.15s) run against a budget whose + total is far smaller (0.02s) proves there is no outer wrap: if there + still were one, this would return the default/empty shape instead of + the real result.""" + monkeypatch.setattr(settings, "conventions_enabled", True) + + async def _slow_check(*_args: object, **_kwargs: object) -> dict[str, Any]: + await asyncio.sleep(0.15) + return {"findings": [{"file": "x.py", "line": 1}], "could_not_run": False} + + git_svc = AsyncMock() + git_svc.conventions_check_for_task.side_effect = _slow_check + c, _qa_id, _task_id = _qa_harness(git_svc) + cc: Any = c + + gaps: list[str] = [] + result = await cc._qa_convention_findings( + uuid4(), MagicMock(), timeout=0.02, gaps=gaps + ) + assert result == [{"file": "x.py", "line": 1}] + assert gaps == [] + + +@pytest.mark.asyncio +async def test_claim_review_normal_path_has_no_evidence_gaps( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Byte-for-byte unchanged normal path: no evidence_gaps key at all when + nothing times out.""" + monkeypatch.setattr(settings, "conventions_enabled", False) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + c, qa_id, task_id = _qa_harness(git_svc) + + env = await c.claim_review(qa_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" not in ev + + +# --------------------------------------------------------------------------- +# claim_doc_task (doc.py) +# --------------------------------------------------------------------------- + + +def _doc_task(task_id: Any, branch: str) -> MagicMock: + return MagicMock( + id=task_id, + status="awaiting_documentation", + assigned_to=None, + task_type="documentation", + team="backend", + branch_name=branch, + quick_context=None, + documents=[], + commits=[{"sha": "abc123", "message": "[x] work"}], + pr_number=7, + pr_url="https://github.com/x/y/pull/7", + dev_notes="done", + acceptance_criteria_status=[], + work_session_id=uuid4(), + ) + + +def _doc_harness(git_svc: AsyncMock) -> tuple[Choreographer, Any, Any]: + doc_id = uuid4() + task_id = uuid4() + branch = "feature/backend/root1234--cellpm56--dev78901" + t_initial = _doc_task(task_id, branch) + t_claimed = MagicMock(**{**t_initial.__dict__, "assigned_to": doc_id}) + + task_svc = AsyncMock() + task_svc.get.return_value = t_initial + task_svc.agent_for.return_value = MagicMock(role="documenter", team="backend") + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.doc_claim.return_value = t_claimed + _stub_empty_ledger(task_svc.session) + + deps = _make_deps(task=task_svc, git=git_svc) + return Choreographer(deps), doc_id, task_id + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_claim_doc_task_diff_timeout_degrades_with_gap(exc: Exception) -> None: + git_svc = AsyncMock() + git_svc.diff.side_effect = exc + git_svc.list_changed_files.return_value = ["README.md"] + c, doc_id, task_id = _doc_harness(git_svc) + + env = await c.claim_doc_task(doc_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" in ev + assert any("pr diff unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_claim_doc_task_checkout_timeout_degrades_with_gap( + exc: Exception, +) -> None: + """The checkout leg (run before evidence assembly) also degrades bounded + instead of an unbounded suppress(Exception) — its gap folds into the + same evidence_gaps list the diff/list_changed_files legs use, drawing + from the SAME shared LegBudget.""" + git_svc = AsyncMock() + git_svc.checkout_branch_in_agent_workspace.side_effect = exc + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + c, doc_id, task_id = _doc_harness(git_svc) + + env = await c.claim_doc_task(doc_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + # The other legs are untouched by the checkout's own timeout. + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" in ev + assert any("workspace checkout unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +async def test_claim_doc_task_normal_path_has_no_evidence_gaps() -> None: + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + c, doc_id, task_id = _doc_harness(git_svc) + + env = await c.claim_doc_task(doc_id, task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" not in ev + + +# --------------------------------------------------------------------------- +# claim_gate_review (pr_gate.py) +# --------------------------------------------------------------------------- + + +def _gate_task() -> MagicMock: + return MagicMock( + id=uuid4(), + status="awaiting_pr_review", + assigned_to=uuid4(), + parent_task_id=None, + task_type="planning", + dependency_ids=[], + team="main_pm", + pr_number=139, + pr_url="https://example/pr/139", + branch_name="feature/main_pm/root", + batch_id=None, + description=None, + acceptance_criteria=[], + ) + + +def _gate_harness(git_svc: AsyncMock) -> tuple[Choreographer, Any, Any]: + task_svc = AsyncMock() + t = _gate_task() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock( + role="pr_reviewer", slug="be-pr-reviewer" + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.unmet_dependency_ids = AsyncMock(return_value=[]) + task_svc.has_earlier_incomplete_code_sibling.return_value = False + task_svc.pr_gate_claim = AsyncMock(return_value=t) + _stub_empty_ledger(task_svc.session) + deps = _make_deps(task=task_svc, git=git_svc) + return Choreographer(deps), t, uuid4() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_claim_gate_review_diff_timeout_degrades_with_gap(exc: Exception) -> None: + git_svc = AsyncMock() + git_svc.diff.side_effect = exc + git_svc.list_changed_files.return_value = ["README.md"] + c, t, reviewer_id = _gate_harness(git_svc) + + env = await c.claim_gate_review(reviewer_id, t.id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff"] == "" + assert "evidence_gaps" in ev + assert any("pr diff unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +async def test_claim_gate_review_files_changed_timeout_degrades_with_gap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """list_changed_files hanging must not sink the whole gate claim, and the + diff leg (which succeeded) must remain intact in the evidence. + + ``_gate_changed_files`` has its own internal ``except Exception`` (an + existing hard-failure fail-open, untouched by this fix) that would + swallow a synchronously-raised exception (of either timeout shape) + before the outer ``run_bounded_leg`` ever saw it — so this uses a + genuinely slow coroutine + a monkeypatched short budget to exercise the + real cancel-at-the-wall path (``asyncio.wait_for`` cancelling the + awaited task via ``CancelledError``, which that ``except Exception`` + does NOT catch), matching what a real hang does in production. + """ + monkeypatch.setattr(settings, "evidence_assembly_timeout_seconds", 0.02) + + async def _hangs(*_args: object, **_kwargs: object) -> list[str]: + await asyncio.sleep(5) + return ["should-not-be-reached"] + + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.side_effect = _hangs + c, t, reviewer_id = _gate_harness(git_svc) + + env = await c.claim_gate_review(reviewer_id, t.id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff"] == "diff content" + assert "evidence_gaps" in ev + assert any("files_changed unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +async def test_claim_gate_review_normal_path_has_no_evidence_gaps() -> None: + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + c, t, reviewer_id = _gate_harness(git_svc) + + env = await c.claim_gate_review(reviewer_id, t.id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff"] == "diff content" + assert "evidence_gaps" not in ev + + +# --------------------------------------------------------------------------- +# _build_i_am_done_ok (i_am_done's success-envelope evidence). Runs strictly +# AFTER the composed transition already committed — advisory, not gating — +# so its list_changed_files leg is bounded exactly like the claim paths. +# --------------------------------------------------------------------------- + + +def _done_task(task_id: Any) -> MagicMock: + return MagicMock( + id=task_id, + branch_name="feature/backend/abc", + commits=[{"sha": "abc123", "message": "x"}], + dev_notes="done", + acceptance_criteria_status=[], + pr_number=5, + pr_url="https://github.com/x/y/pull/5", + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_build_i_am_done_ok_files_changed_timeout_degrades_with_gap( + exc: Exception, +) -> None: + """A hung list_changed_files leg in i_am_done's already-committed + success-envelope builder must not hang the dev's response — it degrades + to an empty files_changed and records the gap.""" + agent_id = uuid4() + task_id = uuid4() + t = _done_task(task_id) + task_svc = AsyncMock() + task_svc.agent_for.return_value = MagicMock(role="developer", team="backend") + _stub_empty_ledger(task_svc.session) + git_svc = AsyncMock() + git_svc.list_changed_files.side_effect = exc + deps = _make_deps(task=task_svc, git=git_svc) + c = Choreographer(deps) + + env = await c._build_i_am_done_ok(agent_id, task_id, t) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["files_changed"] == [] + assert "evidence_gaps" in ev + assert any("files_changed unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +async def test_build_i_am_done_ok_normal_path_has_no_evidence_gaps() -> None: + agent_id = uuid4() + task_id = uuid4() + t = _done_task(task_id) + task_svc = AsyncMock() + task_svc.agent_for.return_value = MagicMock(role="developer", team="backend") + _stub_empty_ledger(task_svc.session) + git_svc = AsyncMock() + git_svc.list_changed_files.return_value = ["README.md"] + deps = _make_deps(task=task_svc, git=git_svc) + c = Choreographer(deps) + + env = await c._build_i_am_done_ok(agent_id, task_id, t) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" not in ev diff --git a/tests/unit/gateway/test_evidence_populates_files_changed.py b/tests/unit/gateway/test_evidence_populates_files_changed.py index f940d461..ea11062f 100644 --- a/tests/unit/gateway/test_evidence_populates_files_changed.py +++ b/tests/unit/gateway/test_evidence_populates_files_changed.py @@ -20,6 +20,8 @@ from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest +from roboco.config import settings +from roboco.exceptions import GitTimeoutError from roboco.services.gateway.content_actions import ContentActions, ContentActionsDeps @@ -223,3 +225,143 @@ async def test_evidence_no_branch_skips_git_calls() -> None: assert body["evidence"]["pr_diff_summary"] == "" git_svc.diff.assert_not_awaited() git_svc.list_changed_files.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Bounded advisory-evidence legs: evidence() must not hang on a slow branch +# fetch / diff / list_changed_files leg — it degrades and records a note in +# evidence_gaps instead (same run_bounded_leg treatment as claim_review / +# claim_doc_task / claim_gate_review). Every timeout-shaped test is +# parametrized over both real timeout shapes: asyncio's own +# cancellation-converted TimeoutError, and GitTimeoutError (_run_git's own +# internal subprocess bound — a GitError/RobocoError subclass, NOT a +# TimeoutError subclass, and the most common real-world single-hung-git-call +# shape since it defaults to a SHORTER window than a leg's own budget). +# --------------------------------------------------------------------------- + +_TIMEOUT_EXCEPTIONS = ( + TimeoutError("hung"), + GitTimeoutError("git diff", 30), +) +_TIMEOUT_IDS = ("asyncio_timeout", "git_timeout") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_evidence_diff_timeout_degrades_with_gap(exc: Exception) -> None: + """A hung git.diff must not hang evidence(): it degrades to an empty + diff, records the gap, and list_changed_files (the other leg) still + comes through untouched.""" + agent_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = _task_with_pr(task_id, commits=["abc"]) + git_svc = AsyncMock() + git_svc.diff.side_effect = exc + git_svc.list_changed_files.return_value = ["README.md"] + workspace_svc = AsyncMock() + evidence_repo = AsyncMock() + evidence_repo.journal_highlights_for_task.return_value = [] + + ca = ContentActions( + _deps_for_evidence(task_svc, git_svc, workspace_svc, evidence_repo) + ) + env = await ca.evidence(agent_id=agent_id, task_id=task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" in ev + assert any("pr diff unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exc", _TIMEOUT_EXCEPTIONS, ids=_TIMEOUT_IDS) +async def test_evidence_branch_fetch_timeout_degrades_with_gap(exc: Exception) -> None: + """A hung workspace branch-fetch must not hang evidence() either — the + subsequent diff/list_changed_files legs still run (against whatever the + workspace already has) and the gap is recorded.""" + agent_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = _task_with_pr(task_id, commits=["abc"]) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + workspace_svc = AsyncMock() + workspace_svc.fetch_branch_for_inspection.side_effect = exc + evidence_repo = AsyncMock() + evidence_repo.journal_highlights_for_task.return_value = [] + + ca = ContentActions( + _deps_for_evidence(task_svc, git_svc, workspace_svc, evidence_repo) + ) + env = await ca.evidence(agent_id=agent_id, task_id=task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" in ev + assert any("branch fetch unavailable" in g for g in ev["evidence_gaps"]) + + +@pytest.mark.asyncio +async def test_evidence_branch_fetch_passes_subprocess_timeout_from_budget() -> None: + """The branch-fetch leg passes its own remaining LegBudget share down as + fetch_branch_for_inspection's subprocess_timeout, so a hung fetch + subprocess self-terminates near the leg's own budget instead of + occupying a thread on the shared default executor for up to + workspace_clone_timeout (300s) after evidence() already gave up on it.""" + agent_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = _task_with_pr(task_id, commits=["abc"]) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + workspace_svc = AsyncMock() + workspace_svc.fetch_branch_for_inspection.return_value = None + evidence_repo = AsyncMock() + evidence_repo.journal_highlights_for_task.return_value = [] + + ca = ContentActions( + _deps_for_evidence(task_svc, git_svc, workspace_svc, evidence_repo) + ) + env = await ca.evidence(agent_id=agent_id, task_id=task_id) + assert env.as_dict()["error"] is None + + workspace_svc.fetch_branch_for_inspection.assert_awaited_once() + call_kwargs = workspace_svc.fetch_branch_for_inspection.await_args.kwargs + assert ( + call_kwargs["subprocess_timeout"] <= settings.evidence_assembly_timeout_seconds + ) + assert call_kwargs["subprocess_timeout"] > 0 + + +@pytest.mark.asyncio +async def test_evidence_normal_path_has_no_evidence_gaps() -> None: + """Byte-for-byte unchanged normal path: no evidence_gaps key at all when + nothing times out.""" + agent_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = _task_with_pr(task_id, commits=["abc"]) + git_svc = AsyncMock() + git_svc.diff.return_value = "diff content" + git_svc.list_changed_files.return_value = ["README.md"] + workspace_svc = AsyncMock() + evidence_repo = AsyncMock() + evidence_repo.journal_highlights_for_task.return_value = [] + + ca = ContentActions( + _deps_for_evidence(task_svc, git_svc, workspace_svc, evidence_repo) + ) + env = await ca.evidence(agent_id=agent_id, task_id=task_id) + body = env.as_dict() + assert body["error"] is None, body + ev = body["evidence"] + assert ev["pr_diff_summary"] == "diff content" + assert ev["files_changed"] == ["README.md"] + assert "evidence_gaps" not in ev diff --git a/tests/unit/gateway/test_pm_decision_autorecord.py b/tests/unit/gateway/test_pm_decision_autorecord.py index e79d8f8d..5402e3b0 100644 --- a/tests/unit/gateway/test_pm_decision_autorecord.py +++ b/tests/unit/gateway/test_pm_decision_autorecord.py @@ -7,6 +7,12 @@ gate runs. This removes the dominant stall where a loaded/weak-model PM forgot the separate note(scope='decision') call and looped on a tracing_gap → respawn. The gate itself is unchanged (see test_pm_decision_window.py) — this only ensures a fresh decision exists. + +``_ensure_pm_decision`` returns a ``PmDecisionOutcome`` ("fresh" / "wrote" / +"transient_failure" / "absent") the caller threads into the gate helper that +runs right after (see test_pm_decision_transient_failure.py for the +gate-satisfaction behavior itself) — these tests pin the outcome value for +each branch. """ from __future__ import annotations @@ -54,8 +60,11 @@ async def test_writes_decision_when_none_exists() -> None: journal.latest_decision_at.return_value = None c = Choreographer(_make_deps(journal=journal)) - await c._ensure_pm_decision(agent_id, task_id, "Merging PR #120; all ACs verified") + outcome = await c._ensure_pm_decision( + agent_id, task_id, "Merging PR #120; all ACs verified" + ) + assert outcome == "wrote" journal.write_decision.assert_awaited_once() _args, kwargs = journal.write_decision.call_args assert kwargs["agent_id"] == agent_id @@ -69,8 +78,9 @@ async def test_skips_when_fresh_decision_already_exists() -> None: journal.latest_decision_at.return_value = datetime.now(UTC) - timedelta(seconds=60) c = Choreographer(_make_deps(journal=journal)) - await c._ensure_pm_decision(uuid4(), uuid4(), "rationale text here") + outcome = await c._ensure_pm_decision(uuid4(), uuid4(), "rationale text here") + assert outcome == "fresh" journal.write_decision.assert_not_awaited() @@ -82,8 +92,11 @@ async def test_writes_when_existing_decision_is_stale() -> None: ) c = Choreographer(_make_deps(journal=journal)) - await c._ensure_pm_decision(uuid4(), uuid4(), "fresh rationale around this point") + outcome = await c._ensure_pm_decision( + uuid4(), uuid4(), "fresh rationale around this point" + ) + assert outcome == "wrote" journal.write_decision.assert_awaited_once() @@ -92,21 +105,29 @@ async def test_noop_on_empty_rationale() -> None: journal = AsyncMock() c = Choreographer(_make_deps(journal=journal)) - await c._ensure_pm_decision(uuid4(), uuid4(), " ") - await c._ensure_pm_decision(uuid4(), uuid4(), None) + outcome_blank = await c._ensure_pm_decision(uuid4(), uuid4(), " ") + outcome_none = await c._ensure_pm_decision(uuid4(), uuid4(), None) + assert outcome_blank == "absent" + assert outcome_none == "absent" journal.latest_decision_at.assert_not_awaited() journal.write_decision.assert_not_awaited() @pytest.mark.asyncio async def test_swallows_write_failure_best_effort() -> None: - """A journal write failure must not crash the verb — the gate then - rejects normally (the pre-fix behaviour), never a 500.""" + """A journal write failure must not crash the verb. Round-2 fix: the + outcome is "transient_failure" (not swallowed into a bare None) so the + caller's gate can treat a DB hiccup as satisfied by the rationale + already in hand — see test_pm_decision_transient_failure.py for the + gate-satisfaction behavior itself.""" journal = AsyncMock() journal.latest_decision_at.return_value = None journal.write_decision.side_effect = RuntimeError("db down") c = Choreographer(_make_deps(journal=journal)) # Must not raise. - await c._ensure_pm_decision(uuid4(), uuid4(), "rationale that triggers a write") + outcome = await c._ensure_pm_decision( + uuid4(), uuid4(), "rationale that triggers a write" + ) + assert outcome == "transient_failure" diff --git a/tests/unit/gateway/test_pm_decision_transient_failure.py b/tests/unit/gateway/test_pm_decision_transient_failure.py new file mode 100644 index 00000000..045baa38 --- /dev/null +++ b/tests/unit/gateway/test_pm_decision_transient_failure.py @@ -0,0 +1,285 @@ +"""PM-decision write-then-gate: transient DB failure must not launder into a +durable rejection/block. + +Live bug: every PM verb (complete / submit_up / submit_root / unblock / +escalate_up / escalate_to_ceo / delegate) runs ``_ensure_pm_decision`` to +auto-record its own rationale as a journal:decision before the freshness +gate (``_check_pm_decision_required`` / ``_check_complete_gates`` / +``_check_submit_up_gates``) runs. The write can lock-timeout under DB +contention (a concurrent claim holding the task row's FK share lock); the +old contract swallowed that and let the gate reject a "missing decision" +the PM's own rationale already answered — a PM retries, keeps getting +rejected while contention lasts, then escalates, and the task ends up +BLOCKED. Transient congestion laundered into a durable blocked task. + +Fix: ``_ensure_pm_decision`` returns a ``PmDecisionOutcome`` — "fresh" / +"wrote" / "transient_failure" / "absent". Every gate helper accepts +``pm_decision_outcome`` (default ``None`` = legacy behavior unchanged) and +treats "transient_failure" as gate-satisfied for THIS call — the rationale +is in the verb payload; the write was only ever a convenience. "absent" (no +rationale at all) still rejects exactly as before. + +These tests exercise the gate helpers directly (the leanest harness that +reaches the actual decision-point) for precise, fast coverage of the +mechanism, plus one near-real-call-site test per verb family +(``_cell_pm_complete_guard``, ``escalate_up``) proving the real wiring. +""" + +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.services.gateway.choreographer import Choreographer, ChoreographerDeps +from roboco.services.gateway.choreographer import _impl as _impl_module +from sqlalchemy.exc import OperationalError + + +def _make_choreographer(**overrides: Any) -> Choreographer: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + return Choreographer(ChoreographerDeps(**base)) + + +_TRANSIENT_WARNING_SUBSTRING = "gate satisfied by verb rationale" + + +# --------------------------------------------------------------------------- +# _check_pm_decision_required (unblock / escalate_up / escalate_to_ceo / delegate) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_check_pm_decision_required_transient_failure_satisfies_gate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A "transient_failure" outcome satisfies the gate for this call even + though no fresh decision exists — the verb's own rationale is the + substance, the write was a convenience. + + ``_impl.py`` logs via ``structlog.get_logger()`` directly; absent this + process having called ``roboco.logging.setup_logging()`` (never true in + a bare unit-test run), structlog uses its own default global config and + never touches stdlib ``logging`` — so ``caplog`` cannot see it. Patching + the module-level ``logger`` object is the reliable way to assert a + structlog call in this harness. + """ + mock_logger = MagicMock() + monkeypatch.setattr(_impl_module, "logger", mock_logger) + journal_svc = AsyncMock() + journal_svc.latest_decision_at.return_value = None # no fresh decision + c = _make_choreographer(journal=journal_svc) + t = MagicMock(id=uuid4()) + + env = await c._check_pm_decision_required( + "unblock", + uuid4(), + t.id, + t, + pm_decision_outcome="transient_failure", + ) + assert env is None + mock_logger.warning.assert_called_once() + call = mock_logger.warning.call_args + assert _TRANSIENT_WARNING_SUBSTRING in call.args[0] + assert call.kwargs.get("verb") == "unblock" + + +@pytest.mark.asyncio +async def test_check_pm_decision_required_absent_still_rejects() -> None: + """ "absent" (no rationale, no fresh decision) rejects exactly as before + — defense-in-depth unchanged.""" + journal_svc = AsyncMock() + journal_svc.latest_decision_at.return_value = None + c = _make_choreographer(journal=journal_svc) + t = MagicMock(id=uuid4()) + + env = await c._check_pm_decision_required( + "unblock", uuid4(), t.id, t, pm_decision_outcome="absent" + ) + assert env is not None + assert env.as_dict()["error"] == "tracing_gap" + + +@pytest.mark.asyncio +async def test_check_pm_decision_required_none_default_unchanged() -> None: + """Every call site that hasn't threaded the outcome through (there are + none left in production, but the param defaults to None for legacy + parity) behaves byte-for-byte as before: no fresh decision rejects.""" + journal_svc = AsyncMock() + journal_svc.latest_decision_at.return_value = None + c = _make_choreographer(journal=journal_svc) + t = MagicMock(id=uuid4()) + + env = await c._check_pm_decision_required("unblock", uuid4(), t.id, t) + assert env is not None + assert env.as_dict()["error"] == "tracing_gap" + + +@pytest.mark.asyncio +async def test_check_pm_decision_required_fresh_path_unchanged() -> None: + """A genuinely fresh decision passes regardless of pm_decision_outcome + — "fresh" is not a special-case bypass, it's the ordinary passing path.""" + journal_svc = AsyncMock() + journal_svc.latest_decision_at.return_value = datetime.now(UTC) + c = _make_choreographer(journal=journal_svc) + t = MagicMock(id=uuid4()) + + env = await c._check_pm_decision_required( + "unblock", uuid4(), t.id, t, pm_decision_outcome="fresh" + ) + assert env is None + + +# --------------------------------------------------------------------------- +# _check_complete_gates (cell_pm / main_pm complete) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_check_complete_gates_transient_failure_satisfies_gate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """See ``test_check_pm_decision_required_transient_failure_satisfies_gate`` + for why the module logger is patched directly instead of using caplog.""" + mock_logger = MagicMock() + monkeypatch.setattr(_impl_module, "logger", mock_logger) + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + journal_svc.has_reflect_for_task.return_value = False + c = _make_choreographer(journal=journal_svc) + + env = await c._check_complete_gates( + uuid4(), + uuid4(), + "closing this task: the cell's contribution merged cleanly", + pm_decision_outcome="transient_failure", + ) + assert env is None + mock_logger.warning.assert_called_once() + call = mock_logger.warning.call_args + assert _TRANSIENT_WARNING_SUBSTRING in call.args[0] + assert call.kwargs.get("verb") == "complete" + + +@pytest.mark.asyncio +async def test_check_complete_gates_absent_still_rejects() -> None: + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + journal_svc.has_reflect_for_task.return_value = False + c = _make_choreographer(journal=journal_svc) + + env = await c._check_complete_gates( + uuid4(), + uuid4(), + "closing this task: the cell's contribution merged cleanly", + pm_decision_outcome="absent", + ) + assert env is not None + assert env.as_dict()["error"] == "tracing_gap" + + +@pytest.mark.asyncio +async def test_check_complete_gates_fresh_path_unchanged() -> None: + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + journal_svc.has_reflect_for_task.return_value = True + c = _make_choreographer(journal=journal_svc) + + env = await c._check_complete_gates( + uuid4(), + uuid4(), + "closing this task: the cell's contribution merged cleanly", + pm_decision_outcome="fresh", + ) + assert env is None + + +# --------------------------------------------------------------------------- +# _cell_pm_complete_guard — the real complete() call site, one hop above the +# gate helper, proving the outcome actually reaches it end to end (without +# dragging in the full cell_pm_complete verb's PR-merge/finalize machinery). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_cell_pm_complete_guard_survives_journal_write_lock_timeout() -> None: + """journal.write_decision raising (a lock-timeout under DB contention) + with a substantive ``notes`` rationale in hand must not reject the PM's + complete — the guard clears (returns None) instead of tracing_gap.""" + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + assigned_to=pm_id, + status="awaiting_pm_review", + pr_number=42, + ) + task_svc = AsyncMock() + task_svc.all_subtasks_terminal.return_value = True + task_svc.uncovered_parent_acceptance_criteria.return_value = [] + # _ensure_pm_decision opens a session.begin_nested() savepoint before + # the write raises — an unshaped AsyncMock's auto-attribute return + # doesn't support `async with`, orphaning the mock's internal coroutine + # (AsyncMockMixin._execute_mock_call never awaited). + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + journal_svc.has_reflect_for_task.return_value = False + journal_svc.latest_decision_at.return_value = None + journal_svc.write_decision.side_effect = OperationalError( + "INSERT INTO journal_entries (id, ...) VALUES (...)", + {}, + Exception("canceling statement due to lock timeout"), + ) + c = _make_choreographer(task=task_svc, journal=journal_svc) + + env = await c._cell_pm_complete_guard( + pm_id, task_id, t, "closing this task: the cell's contribution merged cleanly" + ) + + task_svc.session.begin_nested.assert_called() + assert env is None, env.as_dict() if env is not None else None + + +@pytest.mark.asyncio +async def test_cell_pm_complete_guard_empty_notes_still_rejects() -> None: + """No rationale at all (empty notes) AND no fresh decision on record + still rejects — "absent" is not a bypass.""" + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + assigned_to=pm_id, + status="awaiting_pm_review", + pr_number=42, + ) + task_svc = AsyncMock() + task_svc.all_subtasks_terminal.return_value = True + task_svc.uncovered_parent_acceptance_criteria.return_value = [] + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + journal_svc.has_reflect_for_task.return_value = False + journal_svc.latest_decision_at.return_value = None + c = _make_choreographer(task=task_svc, journal=journal_svc) + + env = await c._cell_pm_complete_guard(pm_id, task_id, t, "") + assert env is not None + assert env.as_dict()["error"] == "tracing_gap" diff --git a/tests/unit/models/test_misc_models.py b/tests/unit/models/test_misc_models.py index b5b2a041..2dd3b70c 100644 --- a/tests/unit/models/test_misc_models.py +++ b/tests/unit/models/test_misc_models.py @@ -88,7 +88,6 @@ def test_agent_instance_default_factory_assigns_id() -> None: def test_orchestrator_agent_config_defaults() -> None: - cfg = OrchestratorAgentConfig( agent_id="be-dev-1", blueprint_path=Path("/tmp/blueprint"), diff --git a/tests/unit/services/test_feature_flags.py b/tests/unit/services/test_feature_flags.py index d4e30e30..7ebc9efd 100644 --- a/tests/unit/services/test_feature_flags.py +++ b/tests/unit/services/test_feature_flags.py @@ -49,7 +49,6 @@ async def test_get_bool_parses_and_defaults() -> None: async def test_apply_overrides_stored_flags_only( monkeypatch: pytest.MonkeyPatch, ) -> None: - # Baseline env defaults. monkeypatch.setattr(cfg, "external_pr_enabled", False) monkeypatch.setattr(cfg, "research_enabled", True) @@ -73,7 +72,6 @@ async def test_apply_overrides_stored_flags_only( async def test_effective_values_use_env_default_when_unset( monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.setattr(cfg, "strategy_engine_enabled", True) async def fake_get(_self: SettingsService, _key: str) -> str | None: diff --git a/tests/unit/services/test_git_conventions_check_fail_closed.py b/tests/unit/services/test_git_conventions_check_fail_closed.py index 3fc6b39f..d3dedf72 100644 --- a/tests/unit/services/test_git_conventions_check_fail_closed.py +++ b/tests/unit/services/test_git_conventions_check_fail_closed.py @@ -149,3 +149,80 @@ async def test_validator_timeout_fails_closed_and_reaps( assert "timed out" in (result.get("reason") or "") fake_proc.kill.assert_called_once() fake_proc.wait.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_run_conventions_validator_timeout_override_used_over_hardcoded( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An explicit ``timeout`` kwarg wins over the module-level hardcoded + default — the advisory claim_review path's shorter budget must actually + reach the subprocess wait, not the fail-closed 120s cap. Sets the module + constant to something LONG (would never fire in the test's real time + budget) so a failure here would prove the override was ignored, not a + coincidence of both values being short.""" + fake_proc = MagicMock() + fake_proc.returncode = None + + async def _communicate() -> tuple[bytes, bytes]: + await asyncio.sleep(30) + return (b"", b"") + + fake_proc.communicate = _communicate + fake_proc.kill = MagicMock() + fake_proc.wait = AsyncMock(return_value=-9) + + async def _fake_exec(*_args: object, **_kwargs: object) -> object: + return fake_proc + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_exec) + monkeypatch.setattr(git_module, "_CONVENTIONS_VALIDATOR_TIMEOUT_SECONDS", 300) + + svc = _service() + result = await svc._run_conventions_validator(tmp_path, ["a.py"], timeout=0.01) + assert result["could_not_run"] is True + assert "timed out after 0.01s" in (result.get("reason") or "") + + +def _task_with_id(branch_name: str) -> MagicMock: + """Like ``_task`` but with a real UUID id — ``conventions_check_for_task`` + calls ``require_uuid(task.id)`` outside the resolution try/except, so a + bare MagicMock id would raise before reaching the validator call.""" + return MagicMock(branch_name=branch_name, id=uuid4()) + + +@pytest.mark.asyncio +async def test_conventions_check_for_task_forwards_timeout_override() -> None: + """``conventions_check_for_task``'s ``timeout`` kwarg must reach + ``_run_conventions_validator`` — the seam ``claim_review`` uses to pin + the ADVISORY (shorter) budget instead of the fail-closed default.""" + svc = _service() + _bind(svc, "_workspace_for_branch", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "list_changed_files", AsyncMock(return_value=["a.py"])) + _bind(svc, "_worktree_for_task", MagicMock(return_value=Path("/tmp/wt"))) + _bind(svc, "_ensure_worktree_for_commit", AsyncMock(return_value=None)) + validator = AsyncMock(return_value={"findings": [], "could_not_run": False}) + _bind(svc, "_run_conventions_validator", validator) + + await svc.conventions_check_for_task( + uuid4(), _task_with_id("feature/backend/abc"), timeout=30.0 + ) + validator.assert_awaited_once_with(Path("/tmp/wt"), ["a.py"], timeout=30.0) + + +@pytest.mark.asyncio +async def test_conventions_check_for_task_default_timeout_is_none() -> None: + """The fail-closed callers (i_am_done's ``_conventions_gate``, pr_pass's + ``_conventions_guard``) never pass ``timeout`` — confirming the default + forwards ``None`` so ``_run_conventions_validator`` falls back to its + hardcoded fail-closed cap, unchanged.""" + svc = _service() + _bind(svc, "_workspace_for_branch", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "list_changed_files", AsyncMock(return_value=["a.py"])) + _bind(svc, "_worktree_for_task", MagicMock(return_value=Path("/tmp/wt"))) + _bind(svc, "_ensure_worktree_for_commit", AsyncMock(return_value=None)) + validator = AsyncMock(return_value={"findings": [], "could_not_run": False}) + _bind(svc, "_run_conventions_validator", validator) + + await svc.conventions_check_for_task(uuid4(), _task_with_id("feature/backend/abc")) + validator.assert_awaited_once_with(Path("/tmp/wt"), ["a.py"], timeout=None) diff --git a/tests/unit/services/test_git_worktree_routing_gaps.py b/tests/unit/services/test_git_worktree_routing_gaps.py index 07aee884..136bc13c 100644 --- a/tests/unit/services/test_git_worktree_routing_gaps.py +++ b/tests/unit/services/test_git_worktree_routing_gaps.py @@ -158,7 +158,7 @@ async def test_conventions_check_runs_validator_in_worktree_not_clone() -> None: captured: list[Path] = [] async def _capture_validator( - workspace: Path, _files: list[str] + workspace: Path, _files: list[str], **_kwargs: object ) -> dict[str, object]: captured.append(Path(workspace)) return {"findings": [], "could_not_run": False} diff --git a/tests/unit/services/test_main_pm_code_guard.py b/tests/unit/services/test_main_pm_code_guard.py index 3606036a..9d0e9b70 100644 --- a/tests/unit/services/test_main_pm_code_guard.py +++ b/tests/unit/services/test_main_pm_code_guard.py @@ -169,7 +169,6 @@ async def test_create_rejects_cell_pm_assignee_plus_code() -> None: @pytest.mark.asyncio async def test_create_allows_cell_pm_assignee_plus_planning() -> None: - be_pm_uuid = AGENTS["be-pm"].uuid svc = TaskService( MagicMock(add=MagicMock(), flush=AsyncMock(), execute=AsyncMock()) diff --git a/tests/unit/services/test_workspace_fetch_branch_for_inspection.py b/tests/unit/services/test_workspace_fetch_branch_for_inspection.py new file mode 100644 index 00000000..b54742bf --- /dev/null +++ b/tests/unit/services/test_workspace_fetch_branch_for_inspection.py @@ -0,0 +1,132 @@ +"""fetch_branch_for_inspection's subprocess_timeout override (adversarial-review +round-2 fix 3): the fetch subprocess must self-bound near the caller's own leg +budget instead of running up to workspace_clone_timeout (300s) on the shared +DEFAULT asyncio executor (asyncio.to_thread, not git.py's dedicated +_GIT_EXECUTOR) after the caller has already given up waiting on it. A +timeout there now raises GitTimeoutError (mirroring _run_git's own +TimeoutExpired -> GitTimeoutError conversion in git.py), not a raw +subprocess.TimeoutExpired, so a bounded caller (run_bounded_leg) catches it +the same way as every other git-touching leg. + +The clone-CREATION step (ensure_workspace, stubbed out here) is untouched by +this fix and keeps its own workspace_clone_timeout (300s) unconditionally — +these tests isolate the FETCH subprocess only. +""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest +from roboco.config import settings +from roboco.exceptions import GitTimeoutError +from roboco.services.workspace import WorkspaceService + +if TYPE_CHECKING: + from pathlib import Path + +# Named constant to satisfy ruff PLR2004 (magic value in comparison). +_EXPECTED_TIMEOUT_SECONDS = 5 + + +def _service() -> WorkspaceService: + session = MagicMock() + session.execute = AsyncMock() + return WorkspaceService(session) + + +def _bind(svc: WorkspaceService, name: str, value: object) -> None: + """Stub `name` on `svc` without tripping mypy's method-assign check.""" + object.__setattr__(svc, name, value) + + +def _wire_resolution(svc: WorkspaceService, workspace: Path) -> None: + """Stub the resolution steps ahead of the fetch subprocess so only the + fetch itself is under test.""" + _bind(svc, "_resolve_branch_to_project_slug", AsyncMock(return_value="roboco")) + _bind(svc, "ensure_workspace", AsyncMock(return_value=workspace)) + + +def _no_project_service_patch() -> Any: + """No git token (project=None skips the decrypt path entirely).""" + return patch( + "roboco.services.project.get_project_service", + return_value=MagicMock(get_by_slug=AsyncMock(return_value=None)), + ) + + +@pytest.mark.asyncio +async def test_default_subprocess_timeout_is_workspace_clone_timeout( + tmp_path: Path, +) -> None: + """Every EXISTING caller omits subprocess_timeout — behavior byte-for-byte + unchanged: the fetch subprocess keeps the 300s workspace_clone_timeout.""" + svc = _service() + _wire_resolution(svc, tmp_path) + captured: list[object] = [] + + def _fake_run(*_args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + captured.append(kwargs.get("timeout")) + return subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="") + + with ( + patch("roboco.services.workspace.subprocess.run", side_effect=_fake_run), + patch("roboco.services.workspace._ensure_agent_owned"), + _no_project_service_patch(), + ): + await svc.fetch_branch_for_inspection(agent_id=uuid4(), branch_name="feature/x") + + assert captured == [settings.workspace_clone_timeout] + + +@pytest.mark.asyncio +async def test_subprocess_timeout_override_reaches_the_fetch(tmp_path: Path) -> None: + svc = _service() + _wire_resolution(svc, tmp_path) + captured: list[object] = [] + + def _fake_run(*_args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + captured.append(kwargs.get("timeout")) + return subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="") + + with ( + patch("roboco.services.workspace.subprocess.run", side_effect=_fake_run), + patch("roboco.services.workspace._ensure_agent_owned"), + _no_project_service_patch(), + ): + await svc.fetch_branch_for_inspection( + agent_id=uuid4(), branch_name="feature/x", subprocess_timeout=12.5 + ) + + assert captured == [12.5] + + +@pytest.mark.asyncio +async def test_timeout_expired_becomes_git_timeout_error(tmp_path: Path) -> None: + """Mirrors _run_git's own TimeoutExpired -> GitTimeoutError conversion + (git.py) so run_bounded_leg catches this uniformly with every other + git-touching leg, instead of a raw subprocess.TimeoutExpired propagating + uncaught to the RobocoError handler.""" + svc = _service() + _wire_resolution(svc, tmp_path) + + def _fake_run(*_args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + timeout = kwargs.get("timeout") + raise subprocess.TimeoutExpired( + cmd="git fetch", + timeout=float(timeout) if isinstance(timeout, int | float) else 0, + ) + + with ( + patch("roboco.services.workspace.subprocess.run", side_effect=_fake_run), + patch("roboco.services.workspace._ensure_agent_owned"), + _no_project_service_patch(), + pytest.raises(GitTimeoutError) as exc_info, + ): + await svc.fetch_branch_for_inspection( + agent_id=uuid4(), branch_name="feature/x", subprocess_timeout=5.0 + ) + assert exc_info.value.timeout == _EXPECTED_TIMEOUT_SECONDS diff --git a/tests/unit/test_agents_config.py b/tests/unit/test_agents_config.py index c52feba1..55c3dd73 100644 --- a/tests/unit/test_agents_config.py +++ b/tests/unit/test_agents_config.py @@ -350,7 +350,7 @@ def test_issue_agent_token_with_ttl_uses_expiring_format( assert payload["id"] == "be-dev-1" assert payload["role"] == "developer" assert payload["team"] == "backend" - assert isinstance(payload["iat"], (int, float)) + assert isinstance(payload["iat"], int | float) assert payload["exp"] == payload["iat"] + 3600