diff --git a/src/backend_server.py b/src/backend_server.py index ff69462..757225c 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -1042,6 +1042,7 @@ def _ensure_schema(conn): schema.migrate_add_planner_review_verdict_event_type(conn) schema.migrate_relax_agent_profile_migrated_from_profile(conn) schema.migrate_drop_work_graph_run_org_node(conn) + schema.migrate_add_concurrency_wait_event_type(conn) class _Server(ThreadingHTTPServer): diff --git a/src/schema.py b/src/schema.py index e70c710..44a1250 100644 --- a/src/schema.py +++ b/src/schema.py @@ -282,7 +282,8 @@ CREATE TABLE execution_event ( event_type TEXT NOT NULL CHECK (event_type IN ( 'task_status_changed', 'node_status_changed', 'artifact_created', 'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated', - 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict' + 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict', + 'concurrency_wait' )), payload TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP @@ -818,6 +819,47 @@ def migrate_add_planner_review_verdict_event_type(conn): conn.commit() +def migrate_add_concurrency_wait_event_type(conn): + """2026-08-13 (real incident, task 71868db6): adds 'concurrency_wait' to + execution_event.event_type's CHECK enum on an ALREADY-POPULATED database + -- same create-copy-drop-rename dance as migrate_add_planner_review_ + verdict_event_type, one accumulated step later. task_execution.py:: + _mark_blocked() now logs this event type when + org_node_lifecycle.ConcurrencyLimitReachedError fires, instead of + routing max_concurrent_goals backpressure through blocked_pending_human/ + execution_error like a real failure. Idempotent and safe to call + unconditionally, including against a freshly create_schema()'d database + (which already has the expanded constraint).""" + row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone() + if row is None or "'concurrency_wait'" in row[0]: + return + conn.executescript( + """ + CREATE TABLE execution_event_new ( + event_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL REFERENCES project(project_id), + task_id TEXT REFERENCES task(task_id), + run_id TEXT REFERENCES work_graph_run(run_id), + node_exec_id TEXT REFERENCES work_node_execution(node_exec_id), + event_type TEXT NOT NULL CHECK (event_type IN ( + 'task_status_changed', 'node_status_changed', 'artifact_created', + 'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated', + 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict', + 'concurrency_wait' + )), + payload TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + INSERT INTO execution_event_new SELECT * FROM execution_event; + DROP TABLE execution_event; + ALTER TABLE execution_event_new RENAME TO execution_event; + CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at); + CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at); + """ + ) + conn.commit() + + def migrate_drop_work_graph_run_org_node(conn): """Org Graph unification Phase 2c §2.1/§2.2 (2026-08-12, user-confirmed 2026-08-08): drops work_graph_run.org_node on an ALREADY-POPULATED diff --git a/src/task_execution.py b/src/task_execution.py index 0b884d8..c15dd46 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -26,7 +26,9 @@ auto-trigger, just invoked by an explicit user action instead of firing automatically the instant a task is created. """ +import json import threading +import uuid from src import e2e_pipeline, org_node_lifecycle, schema, scheduler @@ -220,6 +222,31 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): def _mark_blocked(conn, task_id, exc): + """2026-08-13 (real incident, task 71868db6): ConcurrencyLimitReachedError + is NOT an error -- max_concurrent_goals backpressure is normal, expected + behavior (this dispatch attempt just found no free slot). Routing it + through the same blocked_pending_human/execution_error bucket as a real + failure was itself a bug -- it forced a task that should just wait its + turn into looking like it needs human judgment (see + docs/todo/boss-agent-autonomous-blocked-task-arbiter-requirements.md + FR-7, which documents this as one of several real reasons the taxonomy + needed sharpening). org_node_lifecycle._check_concurrency() raises + BEFORE any task.status transition happens for this attempt (checked + first thing inside trigger_org_node_full, before the real subprocess + call), so task.status is still whatever it was on entry (normally + 'queued' for a fresh dispatch) -- leaving it untouched here means the + next new_goal_dispatch tick naturally retries it, no human/Boss Agent + involvement needed. Still logged as its own execution_event (not + silently invisible) so the concurrency wait is auditable.""" + if isinstance(exc, org_node_lifecycle.ConcurrencyLimitReachedError): + project_id = conn.execute("SELECT project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()[0] + conn.execute( + "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " + "VALUES (?, ?, ?, 'concurrency_wait', ?)", + (str(uuid.uuid4()), project_id, task_id, json.dumps({"detail": str(exc)})), + ) + conn.commit() + return reason = scheduler._blocked_reason_for(exc) if isinstance(exc, scheduler._ROUTABLE_EXCEPTIONS) else "execution_error" scheduler._block_task(conn, task_id, reason, detail=str(exc)) diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 6a2578f..919428c 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -659,6 +659,34 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): # the real exception message must still be recoverable somewhere. self.assertEqual(json.loads(payload)["detail"], "boom") + def test_concurrency_limit_reached_does_not_mark_blocked(self): + """Real incident (2026-08-13, task 71868db6): max_concurrent_goals + backpressure got routed through the same blocked_pending_human/ + execution_error bucket as a genuine failure -- wrong, since it's + normal "no free slot right now" behavior, not an error needing + human judgment. org_node_lifecycle._check_concurrency() raises + before any status transition happens, so task.status must stay + untouched (still 'queued') -- the next new_goal_dispatch tick + retries it -- while the concurrency wait itself is still logged as + a real execution_event for visibility.""" + conn = schema.connect(self.db_path) + conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-2', ?, 'g', 'goal', 'queued', 'planner')", (self.project_id,), + ) + conn.commit() + + exc = org_node_lifecycle.ConcurrencyLimitReachedError("max_concurrent_goals=3 reached (4 active)") + task_execution._mark_blocked(conn, "task-2", exc) + + row = conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-2'").fetchone() + self.assertEqual(row, ("queued", None)) + payload = conn.execute( + "SELECT payload FROM execution_event WHERE task_id = 'task-2' AND event_type = 'concurrency_wait'" + ).fetchone()[0] + self.assertIn("max_concurrent_goals=3", json.loads(payload)["detail"]) + conn.close() + if __name__ == "__main__": unittest.main()