External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

[refactor] reduce xenon C-rank blocks to A (behavior-preserving)

Browse Source 
Extract helpers / flatten conditionals in 11 blocks that rated C(11)+
under xenon --max-absolute B, dropping pr_gate.py module rank B->A in
the process. Pure move-and-call refactors: each extracted helper holds
the original logic verbatim and the caller delegates to it; no control
flow, return values, or side effects changed.

Sites: validators._extract_strs, sequencing.dev_task_collision_edges,
evidence_builder.build_task_handoff, intake_driver._coerce_draft,
task.claim_task_for_agent (2 guards), prompter.create_task_from_draft
(validate+assignee), pr_gate._gate_decision (3 helpers),
orchestrator._handle_stopped_container + _reap_with_service,
_impl._create_subtask_from_inputs + complete.

_impl helper returns tuple[TaskNature, list[str]] to preserve mypy
narrowing of acceptance_criteria at the TaskCreateRequest site.

Also fix vulture: rename unused __aexit__ param tb->_tb in
test_conventions_cache_put.py (was hidden while xenon short-circuited
the gate).
This commit is contained in:
Renn F
2026-06-29 03:03:41 +02:00
parent 0c1c3b345a
commit 3a4a3fe57a
10 changed files with 361 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
roboco/agent_sdk/intake_driver.py
+11 -3
View File
@@ -103,22 +103,30 @@ def _coerce_draft(data: Any) -> dict[str, Any] | None:
if not (isinstance(data, dict) and isinstance(data.get("title"), str)):
return None
coerced = dict(data)
_coerce_spec_fields(coerced, coerce_str_list)
return coerced
def _coerce_spec_fields(
coerced: dict[str, Any], coerce: Callable[[Any], list[str]]
) -> None:
"""Coerce the list-shaped spec fields in place: wrap the_work, flatten the
string-list fields, and flatten each the_work unit's items to ``list[str]``."""
if "the_work" in coerced:
coerced["the_work"] = _coerce_to_list(coerced["the_work"])
for key in _STR_LIST_FIELDS:
if key in coerced:
coerced[key] = coerce_str_list(coerced[key])
coerced[key] = coerce(coerced[key])
work = coerced.get("the_work")
if isinstance(work, list):
coerced["the_work"] = [
{
**unit,
"items": coerce_str_list(unit.get("items")),
"items": coerce(unit.get("items")),
}
for unit in work
if isinstance(unit, dict)
]
return coerced
def _extract_draft(text: str) -> dict[str, Any] | None:
roboco/foundation/policy/content/validators.py
+10 -7
View File
@@ -116,13 +116,7 @@ def _extract_strs(item: Any) -> list[str]:
stripped = item.strip()
return [stripped] if stripped else []
if isinstance(item, dict):
for key in _TEXT_KEYS:
if key in item:
return _extract_strs(item[key])
# No recognized key: keep any bare string values the dict carries.
return [
str(v).strip() for v in item.values() if isinstance(v, str) and v.strip()
]
return _extract_strs_from_dict(item)
if isinstance(item, list):
out: list[str] = []
for sub in item:
@@ -131,6 +125,15 @@ def _extract_strs(item: Any) -> list[str]:
return []
def _extract_strs_from_dict(item: dict) -> list[str]:
"""Resolve a dict element to strings via the first recognized text key,
else any bare string values it carries (recurses through SDK ``$text`` wrappers)."""
for key in _TEXT_KEYS:
if key in item:
return _extract_strs(item[key])
return [str(v).strip() for v in item.values() if isinstance(v, str) and v.strip()]
def coerce_str_list(value: Any) -> list[str]:
"""Coerce a list-of-strings field to a flat ``list[str]``.
roboco/runtime/orchestrator.py
+78 -67
View File
@@ -5961,6 +5961,54 @@ Start by:
return None
return rc == 0
async def _maybe_park_for_exit_error(
self, agent_id: str, instance: Any, graceful: bool
) -> bool:
"""Park the provider on a session/usage limit or a server overload detected
in the dead container's output, instead of crash-retrying into it. Returns
True when parked (caller returns); False to proceed with normal handling.
The probe-resume loop revives the task when the limit lifts / overload clears.
"""
if graceful:
return False
rate_limited_provider = await self._provider_rate_limit_park_target(
agent_id, instance
)
if rate_limited_provider is not None:
logger.warning(
"Session/usage limit detected in agent output; parking provider",
agent_id=agent_id,
provider=rate_limited_provider,
task_id=instance.current_task_id,
)
await self._park_provider_unavailable(
agent_id,
instance,
provider=rate_limited_provider,
retry_after=_RATE_LIMIT_RETRY_AFTER_S,
kind="rate_limited",
)
return True
overloaded_provider = await self._provider_overload_park_target(
agent_id, instance
)
if overloaded_provider is not None:
logger.warning(
"Provider overload detected in agent output; parking provider",
agent_id=agent_id,
provider=overloaded_provider,
task_id=instance.current_task_id,
)
await self._park_provider_unavailable(
agent_id,
instance,
provider=overloaded_provider,
retry_after=_OVERLOAD_RETRY_AFTER_S,
kind="overloaded",
)
return True
return False
async def _handle_stopped_container(
self, agent_id: str, instance: Any, exit_code: int | None
) -> None:
@@ -5990,53 +6038,11 @@ Start by:
await self._park_grok_auth_unavailable(agent_id, instance)
return
graceful = exit_code == 0
# Session/usage-limit parking: the Claude session ("5-hour") limit is a
# 429 the SDK does not retry — the container exits non-zero with a
# 0-token rejection. Detect it in the dead container's output and park
# the provider (instead of crash-respawning straight back into the
# limit); the probe-resume loop revives the task when the quota resets.
if not graceful:
rate_limited_provider = await self._provider_rate_limit_park_target(
agent_id, instance
)
if rate_limited_provider is not None:
logger.warning(
"Session/usage limit detected in agent output; parking provider",
agent_id=agent_id,
provider=rate_limited_provider,
task_id=instance.current_task_id,
)
await self._park_provider_unavailable(
agent_id,
instance,
provider=rate_limited_provider,
retry_after=_RATE_LIMIT_RETRY_AFTER_S,
kind="rate_limited",
)
return
# Server-overload parking: a persistent 529/500/503 from the model API
# kills the run (the SDK already retries transient ones). Detect the
# overload marker in the dead container's output and park the provider —
# the same break as a 429 — instead of crash-retrying into the overload.
if not graceful:
overloaded_provider = await self._provider_overload_park_target(
agent_id, instance
)
if overloaded_provider is not None:
logger.warning(
"Provider overload detected in agent output; parking provider",
agent_id=agent_id,
provider=overloaded_provider,
task_id=instance.current_task_id,
)
await self._park_provider_unavailable(
agent_id,
instance,
provider=overloaded_provider,
retry_after=_OVERLOAD_RETRY_AFTER_S,
kind="overloaded",
)
return
# Park the provider on a session/usage limit or a server overload detected
# in the dead container's output instead of crash-retrying into it. The
# probe-resume loop revives the task when the limit lifts / overload clears.
if await self._maybe_park_for_exit_error(agent_id, instance, graceful):
return
if graceful:
logger.info(
"Agent container exited gracefully",
@@ -8590,6 +8596,26 @@ Start now: evidence(task_id="{task_id}")
grace = settings.gateway_health_grace_seconds
return (now - first_seen).total_seconds() >= grace
async def _should_skip_live_reap(self, t: Any, ts: Any) -> bool:
"""True when a live container should be spared from reaping.
A live container normally protects its task; on a registry MISS (e.g. the
orchestrator restarted and forgot a still-running container) fall back to
asking Docker. A live container is spared UNLESS it is wedged (grok) or
its gateway is broken-but-alive past the grace window — both checks kill +
evict it (returning False here) so the caller falls through to release +
respawn. Short-circuits like the original ``and``: when not live, neither
kill nor recovery check is awaited.
"""
live = self._assignee_has_active_instance(
t
) or await self._assignee_container_running(t)
return (
live
and not await self._maybe_kill_wedged_grok(t, ts)
and not await self._maybe_recover_broken_gateway(t)
)
async def _reap_with_service(self, svc: "TaskService") -> None:
"""Inner reap loop, parameterized by the TaskService to use.
@@ -8606,26 +8632,11 @@ Start now: evidence(task_id="{task_id}")
for t in candidates:
ts = t.last_heartbeat_at
if ts is None or ts < cutoff:
# A live container normally protects its task. Prefer the
# in-memory registry; on a registry MISS (e.g. the orchestrator
# restarted and forgot a still-running container) fall back to
# asking Docker, so we don't reap a task out from under a live
# agent. The sole exception is a wedged GROK container — ACTIVE
# yet firing no verb — which the live skip would shield forever:
# kill + evict it past the grok-idle TTL (then fall through to
# release); a live non-grok agent, or a grok within the TTL, is
# skipped.
live = self._assignee_has_active_instance(
t
) or await self._assignee_container_running(t)
# A live container is spared UNLESS it is wedged (grok) or its
# gateway is broken-but-alive past the grace window — both get
# killed + evicted here so we fall through to release + respawn.
if (
live
and not await self._maybe_kill_wedged_grok(t, ts)
and not await self._maybe_recover_broken_gateway(t)
):
# A live container is spared unless it is wedged (grok) or its
# gateway is broken-but-alive past the grace window — see
# _should_skip_live_reap, which kills + evicts those so we fall
# through to release + respawn.
if await self._should_skip_live_reap(t, ts):
continue
# A provider-parked agent (session-limit / overload / grok-429)
# is OFFLINE with a dead container and a ``rate_limit_lifted``
roboco/services/gateway/choreographer/_impl.py
+56 -40
View File
@@ -61,6 +61,7 @@ if TYPE_CHECKING:
# etc.) via MRO, but ``_LegacyChoreographer`` itself does not inherit
# ``ChoreographerHelpers`` — so mypy can't see those names on ``self`` here.
# The cast below reaches the typed view the mixins use (``_Base`` pattern).
from roboco.models.base import TaskNature
from roboco.services.gateway.choreographer._protocol import ChoreographerHelpers
# Minimum character length enforced on rich_plan["approach"] by the PM
@@ -5136,34 +5137,19 @@ class Choreographer:
message=f"cannot resolve a project for the {inputs.team} subtask",
)
async def _create_subtask_from_inputs(
self,
pm_agent_id: UUID,
parent_task_id: UUID,
parent: Any,
@staticmethod
def _require_subtask_completeness(
inputs: DelegateInputs,
) -> Any:
"""Resolve enums + AGENT_UUIDS slug and call TaskService.create_subtask.
By contract, callers (the `delegate` verb body) MUST run
`_delegate_completeness_check` first, so `inputs.acceptance_criteria`
and `inputs.nature` are guaranteed non-None / non-empty here. The
defensive `TaskCompletenessError` raises preserve correctness if
a future caller bypasses the gateway path — defense-in-depth in
line with the service-layer raise.
) -> tuple[TaskNature, list[str]]:
"""Defensive completeness check (callers run ``_delegate_completeness_check``
first); returns the validated ``TaskNature`` and the non-empty
``acceptance_criteria``. Raises ``TaskCompletenessError`` with field hints
if a non-gateway caller bypassed the gateway check — never silently
substitutes an empty acceptance list.
"""
from roboco.foundation.policy.task_completeness import TaskCompletenessError
from roboco.models.base import TaskNature
from roboco.models.task import TaskCreateRequest
from roboco.seeds.initial_data import AGENT_UUIDS
team_enum, type_enum, complexity_enum = self._resolve_delegate_enums(inputs)
assignee_id = UUID(AGENT_UUIDS[inputs.assigned_to])
# The `or []` collapse was removed. The gateway runs
# `_delegate_completeness_check` BEFORE this helper, so empty/None
# acceptance_criteria here means a non-gateway caller bypassed the
# check. Raise so the service-layer raise can attach the
# field hints — never silently substitute.
if not inputs.acceptance_criteria:
raise TaskCompletenessError(
missing=["acceptance_criteria"],
@@ -5180,9 +5166,7 @@ class Choreographer:
if inputs.nature is None:
raise TaskCompletenessError(
missing=["nature"],
field_hints={
"nature": "one of: technical | non_technical",
},
field_hints={"nature": "one of: technical | non_technical"},
message=(
"_create_subtask_from_inputs called with no nature — "
"completeness check must run first"
@@ -5196,11 +5180,39 @@ class Choreographer:
field_hints={"nature": "one of: technical | non_technical"},
message=f"invalid nature {inputs.nature!r}: {exc}",
) from exc
return nature_enum, inputs.acceptance_criteria
async def _create_subtask_from_inputs(
self,
pm_agent_id: UUID,
parent_task_id: UUID,
parent: Any,
inputs: DelegateInputs,
) -> Any:
"""Resolve enums + AGENT_UUIDS slug and call TaskService.create_subtask.
By contract, callers (the `delegate` verb body) MUST run
`_delegate_completeness_check` first, so `inputs.acceptance_criteria`
and `inputs.nature` are guaranteed non-None / non-empty here. The
defensive `TaskCompletenessError` raises preserve correctness if
a future caller bypasses the gateway path — defense-in-depth in
line with the service-layer raise.
"""
from roboco.models.task import TaskCreateRequest
from roboco.seeds.initial_data import AGENT_UUIDS
team_enum, type_enum, complexity_enum = self._resolve_delegate_enums(inputs)
assignee_id = UUID(AGENT_UUIDS[inputs.assigned_to])
# The gateway runs `_delegate_completeness_check` BEFORE this helper, so
# empty/None acceptance_criteria or nature here means a non-gateway caller
# bypassed the check — _require_subtask_completeness raises with field
# hints rather than silently substituting.
nature_enum, acceptance_criteria = self._require_subtask_completeness(inputs)
resolved_project_id = await self._resolve_subtask_project(parent, inputs)
req = TaskCreateRequest(
title=inputs.title,
description=inputs.description,
acceptance_criteria=inputs.acceptance_criteria,
acceptance_criteria=acceptance_criteria,
parent_ac_refs=inputs.covers_parent_criteria or [],
team=team_enum,
created_by=pm_agent_id,
@@ -6542,6 +6554,18 @@ class Choreographer:
context_briefing=await self._briefing_for(main_pm_agent_id, root_task_id),
).with_introspection(task=t, role="main_pm")
@staticmethod
def _is_umbrella_in_progress(t: Any, role_str: str) -> bool:
"""A MegaTask umbrella sits in_progress branchless (no submit_root/pr_pass),
so the complete spec gate (AWAITING_PM_REVIEW only) would reject it before
main_pm_complete's branchless-aware guard runs — this lets it fall through
to main_pm_complete (CEO merges the root PR; no agent touches master)."""
return (
role_str == "main_pm"
and str(t.status) == "in_progress"
and is_batch_umbrella(batch_id=t.batch_id, parent_task_id=t.parent_task_id)
)
async def complete(self, agent_id: UUID, task_id: UUID, notes: str) -> Envelope:
"""Dispatch to cell_pm_complete or main_pm_complete based on agent role.
@@ -6596,19 +6620,11 @@ class Choreographer:
verb="complete",
):
return soup
# A MegaTask umbrella is branchless by design and never goes through
# submit_root / pr_pass, so it sits in in_progress with no branch/PR.
# The ``complete`` action's source_statuses={AWAITING_PM_REVIEW} spec
# gate would reject it before main_pm_complete's branchless-aware guard
# can run; skip the spec gate for an in_progress batch umbrella and fall
# through to main_pm_complete (CEO merges the root PR; no agent touches
# master). Role membership is preserved; main_pm_complete re-checks
# assignment, subtasks-terminal, and the journal:decision gate.
umbrella_in_progress = (
role_str == "main_pm"
and str(t.status) == "in_progress"
and is_batch_umbrella(batch_id=t.batch_id, parent_task_id=t.parent_task_id)
)
# An in_progress batch umbrella is branchless and skips the complete spec
# gate (AWAITING_PM_REVIEW only) to fall through to main_pm_complete —
# see _is_umbrella_in_progress. Role membership is preserved; main_pm_complete
# re-checks assignment, subtasks-terminal, and the journal:decision gate.
umbrella_in_progress = self._is_umbrella_in_progress(t, role_str)
if not umbrella_in_progress:
decision = spec_module.can_invoke_intent(role, "complete", t, spec_ctx)
if not decision.allowed:
roboco/services/gateway/choreographer/pr_gate.py
+74 -63
View File
@@ -226,6 +226,69 @@ class PRGateMixin(_Base):
)
return (t, agent, role_str, briefing, spec_ctx)
async def _record_gate_verdict_for(
self, verb: str, t: Any, notes: str, *, issues: tuple[str, ...]
) -> None:
"""Author the canonical pr_review verdict note before the transition.
On pr_fail also stamp the assembled PR's head SHA so the next submit_root
can structurally refuse to re-submit the unchanged root (the 2026-06-27
infinite pr_fail re-submit loop). Best-effort: a capture failure leaves
head_sha absent and submit_root fails open rather than wedging the PM.
"""
if verb == "pr_fail":
head_sha = await self._capture_pr_head_sha(t)
self._record_gate_verdict(t, verb, notes, issues=issues, head_sha=head_sha)
else:
self._record_gate_verdict(t, verb, notes, issues=issues)
async def _post_gate_review(
self, t: Any, agent: Any, role_str: str, verb: str, notes: str
) -> None:
"""Post the gate verdict to the PR itself (best-effort, after the DB
transition — a GitHub failure must not roll back the gate decision)."""
reviewer_slug = getattr(agent, "slug", None) or role_str
await self._post_gate_review_to_pr(t, verb, reviewer_slug, notes)
async def _deliver_pr_fail_to_owner(
self, t: Any, reviewer_agent_id: UUID, task_id: UUID, notes: str
) -> None:
"""a2a the pr_fail change-requests to the owning PM (best-effort).
The verdict is posted on the PR but never reaches a PM-readable channel
(no a2a, and _briefing_for / build_task_handoff read neither
pr_reviewer_notes nor notes_structured.pr_review), so without this the
owning PM respawned into needs_revision re-submits the same PR blind —
an infinite pr_fail loop (live on 9980d0a0 / PR #138). Mirrors QA's
fail_review a2a. A Main-PM branch-bearing root is assembled cell work the
Main PM can't fix directly, so steer it to re-delegate + wait for
re-assembly rather than re-submit the unchanged root.
"""
if t.assigned_to is None:
return
team = getattr(t, "team", None)
team_value = str(getattr(team, "value", team))
is_main_pm_root = team_value == spec_module.Team.MAIN_PM.value and bool(
getattr(t, "branch_name", None)
)
steer = (
" Assembled cell work failed — re-delegate the fixes to the"
" owning cell PM(s) and wait for re-assembly; do NOT re-submit"
" the root."
if is_main_pm_root
else ""
)
try:
await self.a2a.send(
from_agent=reviewer_agent_id,
to_agent=t.assigned_to,
skill="code_review",
task_id=task_id,
body=f"PR review needs changes. {notes}{steer}",
)
except Exception:
logger.exception("pr_fail a2a to owning PM failed", task_id=str(task_id))
async def _gate_decision(
self,
reviewer_agent_id: UUID,
@@ -253,21 +316,10 @@ class PRGateMixin(_Base):
)
if blocked is not None:
return blocked
# Author the canonical pr_review verdict note BEFORE the transition so
# it is persisted by the same commit (mirrors post_pr_review). This is
# what keeps notes_structured.pr_review in lock-step with the decision —
# a later pr_fail overwrites an earlier pr_pass verdict instead of
# leaving a stale "passed" on a task that was just sent back. On pr_fail
# also stamp the assembled PR's head SHA so the next submit_root can
# structurally refuse to re-submit the unchanged root (the 2026-06-27
# infinite pr_fail re-submit loop). Best-effort: a capture failure
# (no token, no PR, GitHub error) leaves head_sha absent and the
# submit_root gate fails open rather than wedging the PM.
if verb == "pr_fail":
head_sha = await self._capture_pr_head_sha(t)
self._record_gate_verdict(t, verb, notes, issues=issues, head_sha=head_sha)
else:
self._record_gate_verdict(t, verb, notes, issues=issues)
# Author the canonical pr_review verdict note BEFORE the transition so it
# is persisted by the same commit (mirrors post_pr_review) and stays in
# lock-step with the decision (pr_fail overwrites an earlier pr_pass).
await self._record_gate_verdict_for(verb, t, notes, issues=issues)
runner = self._verb_runner()
try:
t = await runner.run_intent(verb, t, agent, spec_ctx)
@@ -304,54 +356,13 @@ class PRGateMixin(_Base):
task_id=task_id,
verb=verb,
)
# Leave the gate verdict on the PR itself so there's a visible trail on
# the very PR the PM (or CEO) merges. Best-effort and AFTER the DB
# transition — a GitHub failure must not roll back the gate decision.
reviewer_slug = getattr(agent, "slug", None) or role_str
await self._post_gate_review_to_pr(t, verb, reviewer_slug, notes)
# Deliver the change-requests to the owner that now has to act on them
# — the cell PM the runner just re-assigned via _revision_pm_for_task.
# The reviewer posts the verdict on the PR itself but that never reaches
# any PM-readable channel (no a2a, and _briefing_for / build_task_handoff
# read neither pr_reviewer_notes nor notes_structured.pr_review). Without
# this the owning PM respawned into needs_revision saw a generic "needs
# revision" with zero concrete issues, concluded nothing to rework, and
# re-submitted the same PR — an infinite pr_fail loop (live on
# 9980d0a0 / PR #138). Mirrors QA's fail_review a2a to the dev (qa.py:671).
# Best-effort: the transition already committed, so a delivery failure
# must not roll the verdict back or 500 the reviewer.
if verb == "pr_fail" and t.assigned_to is not None:
# A Main-PM branch-bearing root is an assembled cell→root / root→master
# PR — coordination, not the Main PM's own code. The rejection is
# about the cells' merged code, which the Main PM cannot fix directly
# (no code verb). Steer the a2a body to re-delegate + wait for
# re-assembly so the PM doesn't re-submit the unchanged root (the
# 2026-06-27 infinite pr_fail loop). The Envelope ``next`` hint makes
# the same steer via _next_hint_pr_fail.
team = getattr(t, "team", None)
team_value = str(getattr(team, "value", team))
is_main_pm_root = team_value == spec_module.Team.MAIN_PM.value and bool(
getattr(t, "branch_name", None)
)
steer = (
" Assembled cell work failed — re-delegate the fixes to the"
" owning cell PM(s) and wait for re-assembly; do NOT re-submit"
" the root."
if is_main_pm_root
else ""
)
try:
await self.a2a.send(
from_agent=reviewer_agent_id,
to_agent=t.assigned_to,
skill="code_review",
task_id=task_id,
body=f"PR review needs changes. {notes}{steer}",
)
except Exception:
logger.exception(
"pr_fail a2a to owning PM failed", task_id=str(task_id)
)
# Post the gate verdict on the PR itself (best-effort, after the DB
# transition — a GitHub failure must not roll back the gate decision).
await self._post_gate_review(t, agent, role_str, verb, notes)
# a2a the pr_fail change-requests to the owning PM (best-effort) — see
# _deliver_pr_fail_to_owner for the rationale and the Main-PM-root steer.
if verb == "pr_fail":
await self._deliver_pr_fail_to_owner(t, reviewer_agent_id, task_id, notes)
return Envelope.ok(
status=str(t.status),
task_id=str(task_id),
roboco/services/gateway/evidence_builder.py
+30 -10
View File
@@ -86,6 +86,27 @@ def _typed(value: Any, expected: type | tuple[type, ...], default: Any) -> Any:
return value if isinstance(value, expected) else default
def _has_prior_work(
commits: list,
acceptance: list,
highlights: list,
pr_number: int | None,
dev_summary: str | None,
completed_deps: list,
pr_review: dict[str, Any] | None,
) -> bool:
"""True when any resumable prior-work signal is present on the task."""
return bool(
commits
or acceptance
or highlights
or pr_number is not None
or dev_summary
or completed_deps
or pr_review is not None
)
def build_task_handoff(
task: Any, journal_highlights: list[dict[str, Any]]
) -> dict[str, Any] | None:
@@ -111,16 +132,15 @@ def build_task_handoff(
# the concrete issues in every PM briefing so a respawned PM doesn't
# re-submit the same PR blind.
pr_review = _extract_pr_review(getattr(task, "notes_structured", None))
has_prior = bool(
commits
or acceptance
or highlights
or pr_number is not None
or dev_summary
or completed_deps
or pr_review is not None
)
if not has_prior:
if not _has_prior_work(
commits,
acceptance,
highlights,
pr_number,
dev_summary,
completed_deps,
pr_review,
):
return None
handoff: dict[str, Any] = {
"pr_number": pr_number,
roboco/services/prompter.py
+51 -38
View File
@@ -193,6 +193,55 @@ class PrompterService:
return Team.BOARD
return Team.MAIN_PM
def _validate_and_coerce_draft(self, draft_data: dict[str, Any]) -> None:
"""Validate title + acceptance criteria, then flatten the list-shaped
fields (acceptance_criteria / what_this_builds / notes / each the_work
unit's items) to ``list[str]`` in place.
Raises ``ValidationError`` (clean 400) for a missing title or empty /
missing acceptance criteria — a malformed draft (e.g. an incomplete
``propose_batch`` item) would otherwise hit a bare ``KeyError`` and
surface as an opaque 500. Coercion runs here too because a draft can
arrive via re-draft / localStorage, not only the intake choke point.
"""
if not draft_data.get("title"):
raise ValidationError(
message="This task draft is missing a title.", field="title"
)
if not draft_data.get("acceptance_criteria"):
raise ValidationError(
message="This task draft is missing acceptance criteria.",
field="acceptance_criteria",
)
draft_data["acceptance_criteria"] = coerce_str_list(
draft_data.get("acceptance_criteria")
)
draft_data["what_this_builds"] = coerce_str_list(
draft_data.get("what_this_builds")
)
draft_data["notes"] = coerce_str_list(draft_data.get("notes"))
for unit in draft_data.get("the_work") or []:
if isinstance(unit, dict):
unit["items"] = coerce_str_list(unit.get("items"))
if not draft_data["acceptance_criteria"]:
raise ValidationError(
message="This task draft is missing acceptance criteria.",
field="acceptance_criteria",
)
def _resolve_draft_assignee(
self, assigned_to: UUID | None, draft_data: dict[str, Any]
) -> UUID | None:
"""Explicit confirm-button assignment wins; else fall back to any assignee
carried on the draft."""
if assigned_to is not None:
return assigned_to
if not draft_data.get("assigned_to"):
return None
with contextlib.suppress(ValueError):
return UUID(str(draft_data["assigned_to"]))
return None
async def create_task_from_draft(
self,
draft_data: dict[str, Any],
@@ -223,40 +272,7 @@ class PrompterService:
draft through to the task so the analyzer's surface is persisted.
"""
place = placement or BatchPlacement()
# A draft must carry a title + acceptance criteria — without this a
# malformed draft (e.g. an agent's incomplete propose_batch item) hits a
# bare KeyError below and surfaces as an opaque 500 instead of a clean,
# actionable 400.
if not draft_data.get("title"):
raise ValidationError(
message="This task draft is missing a title.", field="title"
)
if not draft_data.get("acceptance_criteria"):
raise ValidationError(
message="This task draft is missing acceptance criteria.",
field="acceptance_criteria",
)
# Flatten the string-list fields to list[str]. The agent sometimes emits
# these as XML-ish <item>…</item> elements the SDK parses into dict
# wrappers ({"item": {"$text": "…"}}); left as-is they crash the
# VARCHAR[] insert (asyncpg: "expected str, got dict") and dump str(dict)
# into the rendered description. Coerce here too — a draft can arrive
# via redraft/localStorage, not only the intake choke point.
draft_data["acceptance_criteria"] = coerce_str_list(
draft_data.get("acceptance_criteria")
)
draft_data["what_this_builds"] = coerce_str_list(
draft_data.get("what_this_builds")
)
draft_data["notes"] = coerce_str_list(draft_data.get("notes"))
for unit in draft_data.get("the_work") or []:
if isinstance(unit, dict):
unit["items"] = coerce_str_list(unit.get("items"))
if not draft_data["acceptance_criteria"]:
raise ValidationError(
message="This task draft is missing acceptance criteria.",
field="acceptance_criteria",
)
self._validate_and_coerce_draft(draft_data)
# Recompose the description from the (possibly edited) structured fields —
# the task always carries a freshly-composed, consistent description.
draft_data["description"] = compose_description(draft_data)
@@ -291,10 +307,7 @@ class PrompterService:
# Explicit assignment (from the confirm button) wins; else fall back to
# any assignee carried on the draft. Resolved before team routing — the
# owner decides the team for a product.
resolved_assigned_to: UUID | None = assigned_to
if resolved_assigned_to is None and draft_data.get("assigned_to"):
with contextlib.suppress(ValueError):
resolved_assigned_to = UUID(str(draft_data["assigned_to"]))
resolved_assigned_to = self._resolve_draft_assignee(assigned_to, draft_data)
team = await self._resolve_owning_team(
draft_data,
roboco/services/sequencing.py
+16 -10
View File
@@ -215,6 +215,21 @@ class SequencingService:
_MIN_COLLISION_PAIR = 2
def _surfaced_siblings(siblings: list) -> list:
"""Siblings carrying a collision surface: a project to collide within and at
least one of intends_to_touch / adds_migration / touches_shared."""
return [
s
for s in siblings
if getattr(s, "project_id", None)
and (
getattr(s, "intends_to_touch", None)
or getattr(s, "adds_migration", False)
or getattr(s, "touches_shared", False)
)
]
def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
"""Wire the dev-task collision DAG for a parent's surfaced siblings.
@@ -237,16 +252,7 @@ def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
reverse edge (which would cycle). ``add_dependency`` dedupes, so repeated
wiring is a no-op on already-wired pairs.
"""
surfaced = [
s
for s in siblings
if getattr(s, "project_id", None)
and (
getattr(s, "intends_to_touch", None)
or getattr(s, "adds_migration", False)
or getattr(s, "touches_shared", False)
)
]
surfaced = _surfaced_siblings(siblings)
if len(surfaced) < _MIN_COLLISION_PAIR:
return []
# Stable order across incremental re-runs: priority is set at creation,
roboco/services/task.py
+34 -22
View File
@@ -6481,6 +6481,38 @@ class TaskService(BaseService):
def _task_status_value(task: TaskTable) -> str:
return task.status.value if hasattr(task.status, "value") else str(task.status)
@staticmethod
def _raise_if_self_review(agent: AgentContext, task: TaskTable) -> None:
"""Refuse a QA/Documenter claim of a task it itself developed."""
if agent.role in (AgentRole.QA, AgentRole.DOCUMENTER):
original_dev = extract_original_developer(task)
if original_dev and str(agent.agent_id) == original_dev:
raise UnauthorizedError(
action="claim",
reason=(
"SELF_REVIEW: Cannot claim a task that you developed. "
f"Leave it for another {agent.role.value}."
),
)
@staticmethod
def _raise_if_main_pm_code_claim(
claimant_is_main_pm: bool, task: TaskTable
) -> None:
"""Refuse a Main-PM claim of a code task it would have to execute."""
if (
claimant_is_main_pm
and _task_type_is_code(task.task_type)
and task.status != TaskStatus.NEEDS_REVISION
):
raise UnauthorizedError(
action="claim",
reason=(
"MAIN_PM_NO_CODE: A Main PM coordinates — it does not own a"
" code task. Leave it for a developer; delegate instead."
),
)
async def claim_task_for_agent(
self,
task_id: UUID,
@@ -6497,16 +6529,7 @@ class TaskService(BaseService):
)
# QA / Documenter cannot claim what they themselves developed.
if agent.role in (AgentRole.QA, AgentRole.DOCUMENTER):
original_dev = extract_original_developer(task)
if original_dev and str(agent.agent_id) == original_dev:
raise UnauthorizedError(
action="claim",
reason=(
"SELF_REVIEW: Cannot claim a task that you developed. "
f"Leave it for another {agent.role.value}."
),
)
self._raise_if_self_review(agent, task)
can_assign = permissions.can_perform_task_action(
agent, TaskAction.ASSIGN, task.team
@@ -6549,18 +6572,7 @@ class TaskService(BaseService):
claimant_is_main_pm = await self._is_main_pm_agent(claim_agent_id)
else:
claimant_is_main_pm = agent.role == AgentRole.MAIN_PM
if (
claimant_is_main_pm
and _task_type_is_code(task.task_type)
and task.status != TaskStatus.NEEDS_REVISION
):
raise UnauthorizedError(
action="claim",
reason=(
"MAIN_PM_NO_CODE: A Main PM coordinates — it does not own a"
" code task. Leave it for a developer; delegate instead."
),
)
self._raise_if_main_pm_code_claim(claimant_is_main_pm, task)
claimed = await self.claim(
task_id, claim_agent_id, allow_reassign=allow_reassign
tests/unit/services/test_conventions_cache_put.py
+1 -1
View File
@@ -41,7 +41,7 @@ class _FakeNested:
async def __aenter__(self) -> None:
self._session.savepoint_started += 1
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
async def __aexit__(self, exc_type: Any, exc: Any, _tb: Any) -> None:
if self._session._duplicate:
self._session.savepoint_rolled_back = True
raise IntegrityError(