Fix: human surface lifecycle hardening (#81)

* fix: harden agent-idle, redis loop, git errors, escalation audit

- i_am_idle no longer 500s when auto-pausing a task whose commits are
  stored as dicts: tolerate dict-or-object commit refs and run the
  synthetic-checkpoint computation inside the swallowing try block.
- The stream event loop no longer logs an idle redis read-timeout as an
  ERROR every cycle; the blocking-read timeout is treated as a normal idle.
- Git command failures surface git's own (secret-scrubbed) stderr in the
  error message instead of a bare 'Command failed', so push/fetch
  rejections are diagnosable; the injected PAT is redacted.
- The escalate-to-pool redirect emits the task.pending audit event,
  closing a status mutation that previously skipped the audit log.

* fix: let privileged operators set task status via an audited override

The task update route silently dropped a 'status' field in the request body,
so a CEO/admin could not transition a task wedged in a state with no valid
in-band move (e.g. a blocked task whose work merged out-of-band) — the panel
returned 200 while nothing changed. Add 'status' to the update schema and
apply it through a new audited 'admin_set_status' that bypasses the strict
transition validator but always records the audit event. The override
requires elevated permissions; ordinary field updates are unchanged.

* fix: stop human chat sessions from expiring between messages

Messaging sessions fell back to a hardcoded 300s idle timeout, shorter than a
normal pause in a human conversation: the sweeper closed the session and the
next message opened a new one, so a person could not hold a continuous chat.
Make the idle timeout configurable (session_idle_timeout_seconds, default
3600) and resolve an unset timeout to it at every session-creation path
instead of the 300s column fallback.

* fix: resolve doubled doc paths and stop the indexer warning flood

The doc-path resolver returned absolute paths verbatim, so a documenter path
that doubled the base segment (/app/docs/docs/...) never resolved on disk and
the docs never indexed into RAG. Reduce an absolute path under the docs base
to a relative one before normalizing, leaving truly-external absolute paths
for the indexer to skip. The indexer now skips non-markdown source files and
logs a missing/non-doc source at debug instead of warning on every pass.

* fix: reject project repo URLs that point at a protected repository

Add a configurable denylist (protected_git_urls) enforced in the project
create and update paths, so a project cannot be registered against a
repository that must not receive agent commits or merges (e.g. the roboco
source repo during a smoke run). Empty by default (no behavior change);
operators set it to sandbox smoke-test projects.

* fix: let an agent release a blocked task back to the pool

A developer (or QA/doc) trapped on a 'blocked' task had no legal forward
move — every verb rejected from that state — so the dispatcher kept
respawning it with nothing to do. Allow 'unclaim' to release a blocked task
the agent owns back to pending (assignment cleared, work session abandoned,
audited), so the cell PM can re-delegate it instead of the agent churning.

* fix: keep blocked-dev churn out and cell tasks out of board hands

- The dispatcher no longer respawns the owner of a blocked task: from blocked
  the owner has no legal move, so respawning only churns; it is revived on
  unblock or released via unclaim.
- Escalation no longer hands a cell (backend/frontend/ux_ui) coordination task
  to a board/advisory role — such an escalation is diverted to the cell pool,
  matching the existing executable-task guard. main_pm targets are unaffected.

* chore: add an opt-in full clean-slate to the reset script

FULL_RESET=1 wipes everything under the roboco data root except the
persistent service stores (ollama/postgres/redis) and clears the persisted
agent Claude session dirs (ROBOCO_CLAUDE_STATE_DIRS), which otherwise replay
across runs. Default off — the existing DB/Redis wipe + workspace git-reset
is unchanged.

* ++

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
2026-06-08 01:59:51 +02:00
committed by GitHub
co-authored by Renn F
parent 199169ef0e
commit f58fd4530e
22 changed files with 475 additions and 23 deletions
+23
View File
@@ -515,12 +515,35 @@ async def update_task(
# Transform input data for database storage
updates = transform_update_data(data)
# `status` is not a free-form field — it is an audited admin override so a
# privileged operator can recover a task wedged in a state with no valid
# in-band transition. Pop it out of the generic field update and apply it
# through the audited path, gated on elevated permissions.
new_status = updates.pop("status", None)
task = await service.update(task_id, **updates)
if not task:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Task update failed unexpectedly",
)
if new_status is not None and new_status != task.status:
if not has_higher_perms:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only privileged roles may override task status.",
)
task = await service.admin_set_status(
task_id,
new_status,
actor_id=agent.agent_id,
actor_role=getattr(agent, "role", None),
)
if not task:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Task status override failed unexpectedly",
)
await db.commit()
return task_to_response(task)
+5
View File
@@ -218,6 +218,11 @@ class TaskUpdate(BaseModel):
auditor_notes: str | None = None
quick_context: str | None = None
# Lifecycle override — privileged/admin only. Applied by the route as an
# audited force-transition (so an operator can recover a task wedged in a
# state with no valid in-band move), never as a free-form field set.
status: TaskStatus | None = None
@model_validator(mode="before")
@classmethod
def _reject_explicit_blank_acceptance_criteria(cls, data: Any) -> Any:
+19
View File
@@ -279,6 +279,25 @@ class Settings(BaseSettings):
),
)
session_idle_timeout_seconds: int = Field(
default=3600,
ge=30,
description=(
"Idle seconds before a messaging session is swept closed. The "
"previous 300s default was shorter than a human conversation pause, "
"so a person's chat session expired and reopened between messages."
),
)
protected_git_urls: list[str] = Field(
default_factory=list,
description=(
"Repo URL substrings a project may not point at (e.g. the roboco "
"source repo). Blocks agent commits/merges from reaching a protected "
"repository; set this to sandbox smoke-test projects."
),
)
# ==========================================================================
# Agent Guardrails (per-session budgets, loop detection, SLAs)
# ==========================================================================
+7
View File
@@ -15,6 +15,7 @@ from typing import Any
import redis.asyncio as redis
import structlog
from redis.exceptions import ResponseError
from redis.exceptions import TimeoutError as RedisTimeoutError
from roboco.config import settings
from roboco.models.events import Event, EventType
@@ -219,6 +220,12 @@ class StreamEventBus:
if await self._handle_response_error(e, streams):
continue
await asyncio.sleep(1)
except (RedisTimeoutError, TimeoutError):
# An idle XREADGROUP(block=...) hits the client read-timeout when
# no new message arrives within the block window. This is the
# normal idle path, not an error — re-block on the next iteration
# without logging or back-off.
continue
except Exception as e:
logger.error("Error in stream event loop", error=str(e))
await asyncio.sleep(1)
+29 -3
View File
@@ -5,6 +5,7 @@ Structured exception hierarchy for the AI Agents Company system.
All exceptions include context for debugging and logging.
"""
import re
from typing import Any, ClassVar
from uuid import UUID
@@ -435,16 +436,41 @@ class GitError(ServiceError):
)
def _scrub_git_secrets(text: str) -> str:
"""Redact credentials a git command may echo into stderr.
Push/fetch run with the PAT injected via a URL or an ``http.extraheader``
Basic header; never surface those verbatim in an error message or log.
"""
if not text:
return text
text = re.sub(r"(://)[^/@\s]+@", r"\1***@", text)
text = re.sub(r"(?i)(authorization:\s*basic\s+)\S+", r"\1***", text)
text = re.sub(r"(?i)(extraheader=\S*?basic\s+)\S+", r"\1***", text)
text = re.sub(
r"\b(gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,})\b", "***", text
)
return text
class GitCommandError(GitError):
"""Git command execution failed."""
def __init__(self, command: str, stderr: str) -> None:
scrubbed = _scrub_git_secrets(stderr or "")
# Surface a short, secret-free tail of git's own stderr in the message so
# the real reason (403, non-fast-forward, ...) is visible to callers that
# only render ``.message`` instead of swallowing it as "Command failed".
tail = " ".join(scrubbed.split())[-300:]
message = f"Command failed: {command}"
if tail:
message = f"{message}{tail}"
super().__init__(
message=f"Command failed: {command}",
details={"command": command, "stderr": stderr},
message=message,
details={"command": command, "stderr": scrubbed},
)
self.command = command
self.stderr = stderr
self.stderr = scrubbed
class GitTimeoutError(GitError):
+2 -2
View File
@@ -118,7 +118,7 @@ OLLAMA_ROLE_DEFAULTS: dict[str, str] = {
}
# The Ollama model picked for "pure Ollama" mode's GLOBAL row when the
# caller doesn't override. Kimi K2.6 wins as the generalist because it has
# caller doesn't override. Minimax M3 wins as the generalist because it has
# the strongest reasoning/tool-use profile and can fall back to coding/writing
# adequately if a role ends up mapped to the global default.
OLLAMA_DEFAULT_MODEL: str = "kimi-k2.6:cloud"
OLLAMA_DEFAULT_MODEL: str = "minimax-m3:cloud"
+7 -2
View File
@@ -4808,11 +4808,16 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
async def _handle_dev_existing_owner(
self, task: dict[str, Any], status: str, agent_slug: str
) -> None:
"""Respawn existing dev for needs_revision / in_progress / claimed / blocked."""
"""Respawn existing dev for needs_revision / in_progress / claimed."""
# A `blocked` task is waiting for its blocker to clear (PM / dependency);
# the owner has no legal move from `blocked`, so respawning it does
# nothing but churn. It is revived only when unblocked back to
# in_progress, or released to the pool (unclaim) for re-delegation.
if status == "blocked":
return
if status in (
"in_progress",
"claimed",
"blocked",
) and not self._is_agent_active(agent_slug):
logger.info(
"Respawning agent for orphaned task",
+11 -4
View File
@@ -2707,11 +2707,18 @@ class Choreographer:
Failure is logged and swallowed the pause already happened and the
caller must not be affected by a checkpoint DB error.
"""
commit_refs = [c.sha for c in (task.commits or [])[-3:]]
commit_count = len(task.commits or [])
state_summary = f"auto-paused on i_am_idle (commits: {commit_count})"
remaining_work = commit_refs if commit_refs else ["no commits yet"]
try:
commits = task.commits or []
# commits may be hydrated as CommitRef objects or as plain dicts
# (JSON column round-trip); tolerate both rather than assuming `.sha`.
commit_refs = [
c.get("sha") if isinstance(c, dict) else getattr(c, "sha", None)
for c in commits[-3:]
]
commit_refs = [ref for ref in commit_refs if ref]
commit_count = len(commits)
state_summary = f"auto-paused on i_am_idle (commits: {commit_count})"
remaining_work = commit_refs if commit_refs else ["no commits yet"]
await self.task.add_checkpoint(
task_id=task.id,
agent_id=agent_id,
+21 -3
View File
@@ -20,6 +20,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload
from roboco.config import settings
from roboco.db.tables import (
ChannelTable,
GroupTable,
@@ -378,6 +379,19 @@ class MessagingService(BaseService):
# SESSION OPERATIONS (TASK-015)
# =========================================================================
@staticmethod
def _resolve_session_timeout(requested: int | None) -> int:
"""Resolve a session's idle-timeout, defaulting to the configurable value.
An unset timeout previously fell through to the column default of 300s,
which is shorter than a human conversation pause the session was swept
between messages and a new one opened on the next post. Resolve it to
``session_idle_timeout_seconds`` so human-paced chats stay on one session.
"""
if requested is not None:
return requested
return settings.session_idle_timeout_seconds
async def create_session(self, req: SessionCreateRequest) -> SessionTable:
"""
Create a new session in a group.
@@ -404,7 +418,7 @@ class MessagingService(BaseService):
group_id=req.group_id,
max_message_count=req.max_message_count,
max_content_length=req.max_content_length,
timeout_seconds=req.timeout_seconds,
timeout_seconds=self._resolve_session_timeout(req.timeout_seconds),
status=SessionStatus.ACTIVE,
scope=req.scope,
)
@@ -643,7 +657,7 @@ class MessagingService(BaseService):
max_time_window=(_minutes_to_timedelta(request.max_time_window_minutes)),
max_message_count=request.max_message_count,
max_content_length=request.max_content_length,
timeout_seconds=request.timeout_seconds,
timeout_seconds=self._resolve_session_timeout(request.timeout_seconds),
status=SessionStatus.ACTIVE,
)
self.session.add(new_session)
@@ -1141,7 +1155,11 @@ class MessagingService(BaseService):
group_id=cast("UUID", group.id),
max_message_count=(req.config.max_message_count if req.config else None),
max_content_length=(req.config.max_content_length if req.config else None),
timeout_seconds=(req.config.timeout_seconds if req.config else 300),
timeout_seconds=(
req.config.timeout_seconds
if req.config and req.config.timeout_seconds is not None
else settings.session_idle_timeout_seconds
),
scope=req.scope,
)
+10 -2
View File
@@ -85,8 +85,16 @@ class DocsIndexPlugin(BaseIndexPlugin):
]
return md_files + txt_files
if source_path.exists():
return [source_path]
logger.warning(f"Source not found: {source}")
# Only markdown/text files are docs; a recorded path to a source
# file (e.g. a .tsx) is not indexable and is skipped quietly.
if source_path.suffix.lower() in {".md", ".txt"}:
return [source_path]
logger.debug(f"Skipping non-doc source path: {source}")
return []
# A documenter may record a path for a doc that was not written under
# the docs root; this is not actionable at index time, so log at debug
# instead of flooding a warning on every pass.
logger.debug(f"Doc source not found, skipping: {source}")
return []
def _read_file_record(
+22
View File
@@ -11,7 +11,9 @@ from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from roboco.config import settings
from roboco.db.tables import ProjectTable
from roboco.exceptions import ValidationError
from roboco.models.base import Team
from roboco.models.project import ProjectCreate, ProjectUpdate
from roboco.services.base import BaseService, ConflictError, NotFoundError
@@ -35,6 +37,22 @@ class ProjectService(BaseService):
# CRUD OPERATIONS
# =========================================================================
def _assert_git_url_allowed(self, git_url: str | None) -> None:
"""Reject a project repo URL that matches a protected (denylisted) repo.
Keeps agent commits/merges out of a repository that must not receive
them e.g. the roboco source repo during a smoke run.
"""
if not git_url:
return
for protected in settings.protected_git_urls:
if protected and protected in git_url:
raise ValidationError(
"Project git_url may not point at a protected repository "
f"('{protected}').",
field="git_url",
)
async def create(
self,
data: ProjectCreate,
@@ -61,6 +79,8 @@ class ProjectService(BaseService):
resource_type="project",
)
self._assert_git_url_allowed(data.git_url)
# Encrypt git token if provided
encrypted_token = None
if data.git_token:
@@ -141,6 +161,8 @@ class ProjectService(BaseService):
if not project:
return None
self._assert_git_url_allowed(data.git_url)
# Handle git_token specially (empty string clears, None leaves unchanged)
token_updated = False
if data.git_token is not None:
+119 -4
View File
@@ -87,6 +87,12 @@ _DESCENDANT_EXECUTABLE_TASK_TYPES: frozenset[str] = frozenset(
{TaskType.CODE.value, TaskType.DOCUMENTATION.value, TaskType.DESIGN.value}
)
# Implementation-cell teams. A board/advisory role must never own a cell task —
# including the cell's own coordination/planning task (which carries a cell team
# but not a CODE/DOC/DESIGN type), so escalating one toward a board role is
# diverted to the cell pool instead of handing ownership up.
_CELL_TEAMS: frozenset[str] = frozenset({"backend", "frontend", "ux_ui"})
def _is_descendant_executable_task(task: TaskTable) -> bool:
"""True for a child task that does cell-executed work (#14 guard).
@@ -108,6 +114,19 @@ def _is_descendant_executable_task(task: TaskTable) -> bool:
return str(type_value) in _DESCENDANT_EXECUTABLE_TASK_TYPES
def _is_cell_team_task(task: TaskTable) -> bool:
"""True for a descendant task owned by an implementation cell.
Complements ``_is_descendant_executable_task``: a cell's coordination /
planning task carries a cell ``team`` but not a CODE/DOC/DESIGN ``task_type``,
so the executable-type check alone would let it escalate onto a board role.
"""
if task.parent_task_id is None:
return False
team_value = getattr(task.team, "value", task.team)
return str(team_value) in _CELL_TEAMS
# Notes fields (dev_notes, qa_notes, quick_context) are append-only —
# every revision cycle adds more. Cap total size so a task that cycles
# dozens of times doesn't grow into megabytes. When we exceed the cap,
@@ -974,6 +993,47 @@ class TaskService(BaseService):
)
return task
async def admin_set_status(
self,
task_id: UUID,
new_status: TaskStatus,
*,
actor_id: str | UUID | None = None,
actor_role: str | None = None,
) -> TaskTable | None:
"""Privileged override: set a task's status directly, always audited.
Bypasses the strict transition validator so an operator can recover a
task wedged in a state with no valid in-band move (e.g. a ``blocked``
task whose work already merged out-of-band). The change is recorded in
the audit log like any other transition no status change may skip it.
"""
task = await self.get(task_id)
if not task:
return None
from_status = (
task.status.value
if isinstance(task.status, TaskStatus)
else str(task.status)
)
task.status = new_status
await self.session.flush()
self._emit_status_transition_audit(
task,
from_status=from_status,
to_status=new_status.value,
agent_role=actor_role,
audit_agent_id=actor_id,
)
self.log.info(
"Task status set via admin override",
task_id=str(task_id),
from_status=from_status,
to_status=new_status.value,
actor=str(actor_id) if actor_id else None,
)
return task
async def delete(self, task_id: UUID) -> bool:
"""Delete a task and all its descendants."""
task = await self.get(task_id)
@@ -1735,8 +1795,16 @@ class TaskService(BaseService):
from roboco.services.docs import DOCS_BASE_PATH
path = Path(rel_path)
# An absolute path already rooted at the docs base may double the
# segment (``/app/docs/docs/...``) or simply be re-anchored here; reduce
# it to a path relative to the base so the normalization below applies
# uniformly. An absolute path OUTSIDE the docs root (e.g. a workspace
# source file) is returned as-is for the indexer to skip.
if path.is_absolute():
return str(path)
try:
path = path.relative_to(DOCS_BASE_PATH)
except ValueError:
return str(path)
parts = path.parts
if parts and parts[0] == DOCS_BASE_PATH.name:
path = Path(*parts[1:]) if len(parts) > 1 else Path()
@@ -2238,6 +2306,37 @@ class TaskService(BaseService):
task.active_claimant_id = cast("Any", None)
await self.session.flush()
return task
# A task the agent owns but cannot advance — it is `blocked` (a blocker
# it cannot self-resolve, or a dependency block) — is otherwise a trap:
# from `blocked` the agent has no legal forward verb and the dispatcher
# keeps respawning it. Releasing the claim returns the task to the pool
# for the cell PM to re-delegate. Audited; the active WorkSession is
# abandoned so a re-claim does not trip the uniqueness constraint.
if task.status == TaskStatus.BLOCKED:
pre_status = (
task.status.value
if isinstance(task.status, TaskStatus)
else str(task.status)
)
prior_owner = cast("Any", task.claimed_by or task.assigned_to)
if task.work_session_id:
await self._abandon_work_session_best_effort(
task.work_session_id, reason="agent-unclaim-from-blocked"
)
task.work_session_id = cast("Any", None)
task.status = TaskStatus.PENDING
task.assigned_to = cast("Any", None)
task.claimed_by = cast("Any", None)
task.active_claimant_id = cast("Any", None)
await self.session.flush()
self._emit_status_transition_audit(
task,
from_status=pre_status,
to_status=TaskStatus.PENDING.value,
agent_role=None,
audit_agent_id=prior_owner,
)
return task
if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS):
return None
@@ -3416,9 +3515,9 @@ class TaskService(BaseService):
primitive so both the gateway ``escalate`` verb and the HTTP escalate
route are covered.
"""
if _is_descendant_executable_task(task) and await self._is_board_advisory_agent(
target_agent_id
):
if (
_is_descendant_executable_task(task) or _is_cell_team_task(task)
) and await self._is_board_advisory_agent(target_agent_id):
await self._release_code_task_to_pool(
task=task,
escalator_slug=escalator_slug,
@@ -5656,6 +5755,12 @@ class TaskService(BaseService):
re-dispatch source), and appends an audit note explaining why the board
hand-off was refused.
"""
pre_status = (
task.status.value
if isinstance(task.status, TaskStatus)
else str(task.status)
)
prior_owner = cast("Any", task.claimed_by or task.assigned_to)
task.assigned_to = cast("Any", None)
task.claimed_by = cast("Any", None)
task.active_claimant_id = cast("Any", None)
@@ -5670,6 +5775,16 @@ class TaskService(BaseService):
)
task.dev_notes = existing_notes + note
await self.session.flush()
# This path sets PENDING directly (bypassing the strict transition
# validator), so emit the task.pending audit explicitly — no status
# change may skip the audit log.
self._emit_status_transition_audit(
task,
from_status=pre_status,
to_status=TaskStatus.PENDING.value,
agent_role=None,
audit_agent_id=prior_owner,
)
self.log.info(
"Descendant executable task released to pool instead of board escalation",
task_id=str(task.id),