宜博 / lemongraph

Commit 75162ac63ceddd4a6e07ce1dfc12238fbf60a882

Authored by bboonnee
1 parent 6c4c8f12

fix: concurrency-limit backpressure should not become blocked_pending_human 

Real incident (2026-08-13, task 71868db6): ConcurrencyLimitReachedError isn't an
error -- it just means this dispatch attempt found no free slot, which is normal,
expected backpressure. Routing it through the same blocked_pending_human/
execution_error bucket as a genuine failure was itself a bug: it forced a task
that should just wait its turn into looking like it needs human judgment.

_check_concurrency() raises before any task.status transition happens for this
attempt, so task_execution.py::_mark_blocked() now leaves status untouched
(still 'queued' for a fresh dispatch) instead of blocking -- the next
new_goal_dispatch tick naturally retries it, no human/Boss Agent involvement
needed. Still logged as a new 'concurrency_wait' execution_event (new
migrate_add_concurrency_wait_event_type, same accumulated-CHECK-enum pattern as
the existing event-type migrations) so the wait stays auditable, not silently
invisible.

This is the first concrete instance of FR-7 in
docs/todo/boss-agent-autonomous-blocked-task-arbiter-requirements.md (system
should emit distinguishable blocked_reasons/non-error states at the point of
failure, not dump everything into one generic bucket).

803/803 src tests, 134/134 supervisor tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Inline Side-by-side
Showing 4 changed files with 99 additions and 1 deletions
src/backend_server.py
  View file @75162ac
... ... @@ -1042,6 +1042,7 @@ def _ensure_schema(conn):
1042 1042 schema.migrate_add_planner_review_verdict_event_type(conn)
1043 1043 schema.migrate_relax_agent_profile_migrated_from_profile(conn)
1044 1044 schema.migrate_drop_work_graph_run_org_node(conn)
  1045 + schema.migrate_add_concurrency_wait_event_type(conn)
1045 1046  
1046 1047  
1047 1048 class _Server(ThreadingHTTPServer):
... ...
src/schema.py
  View file @75162ac
... ... @@ -282,7 +282,8 @@ CREATE TABLE execution_event (
282 282 event_type TEXT NOT NULL CHECK (event_type IN (
283 283 'task_status_changed', 'node_status_changed', 'artifact_created',
284 284 'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
285   - 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict'
  285 + 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict',
  286 + 'concurrency_wait'
286 287 )),
287 288 payload TEXT NOT NULL,
288 289 created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
... ... @@ -818,6 +819,47 @@ def migrate_add_planner_review_verdict_event_type(conn):
818 819 conn.commit()
819 820  
820 821  
  822 +def migrate_add_concurrency_wait_event_type(conn):
  823 + """2026-08-13 (real incident, task 71868db6): adds 'concurrency_wait' to
  824 + execution_event.event_type's CHECK enum on an ALREADY-POPULATED database
  825 + -- same create-copy-drop-rename dance as migrate_add_planner_review_
  826 + verdict_event_type, one accumulated step later. task_execution.py::
  827 + _mark_blocked() now logs this event type when
  828 + org_node_lifecycle.ConcurrencyLimitReachedError fires, instead of
  829 + routing max_concurrent_goals backpressure through blocked_pending_human/
  830 + execution_error like a real failure. Idempotent and safe to call
  831 + unconditionally, including against a freshly create_schema()'d database
  832 + (which already has the expanded constraint)."""
  833 + row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone()
  834 + if row is None or "'concurrency_wait'" in row[0]:
  835 + return
  836 + conn.executescript(
  837 + """
  838 + CREATE TABLE execution_event_new (
  839 + event_id TEXT PRIMARY KEY,
  840 + project_id TEXT NOT NULL REFERENCES project(project_id),
  841 + task_id TEXT REFERENCES task(task_id),
  842 + run_id TEXT REFERENCES work_graph_run(run_id),
  843 + node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
  844 + event_type TEXT NOT NULL CHECK (event_type IN (
  845 + 'task_status_changed', 'node_status_changed', 'artifact_created',
  846 + 'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
  847 + 'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict',
  848 + 'concurrency_wait'
  849 + )),
  850 + payload TEXT NOT NULL,
  851 + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
  852 + );
  853 + INSERT INTO execution_event_new SELECT * FROM execution_event;
  854 + DROP TABLE execution_event;
  855 + ALTER TABLE execution_event_new RENAME TO execution_event;
  856 + CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
  857 + CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);
  858 + """
  859 + )
  860 + conn.commit()
  861 +
  862 +
821 863 def migrate_drop_work_graph_run_org_node(conn):
822 864 """Org Graph unification Phase 2c §2.1/§2.2 (2026-08-12, user-confirmed
823 865 2026-08-08): drops work_graph_run.org_node on an ALREADY-POPULATED
... ...
src/task_execution.py
  View file @75162ac
... ... @@ -26,7 +26,9 @@ auto-trigger, just invoked by an explicit user action instead of firing
26 26 automatically the instant a task is created.
27 27 """
28 28  
  29 +import json
29 30 import threading
  31 +import uuid
30 32  
31 33 from src import e2e_pipeline, org_node_lifecycle, schema, scheduler
32 34  
... ... @@ -220,6 +222,31 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
220 222  
221 223  
222 224 def _mark_blocked(conn, task_id, exc):
  225 + """2026-08-13 (real incident, task 71868db6): ConcurrencyLimitReachedError
  226 + is NOT an error -- max_concurrent_goals backpressure is normal, expected
  227 + behavior (this dispatch attempt just found no free slot). Routing it
  228 + through the same blocked_pending_human/execution_error bucket as a real
  229 + failure was itself a bug -- it forced a task that should just wait its
  230 + turn into looking like it needs human judgment (see
  231 + docs/todo/boss-agent-autonomous-blocked-task-arbiter-requirements.md
  232 + FR-7, which documents this as one of several real reasons the taxonomy
  233 + needed sharpening). org_node_lifecycle._check_concurrency() raises
  234 + BEFORE any task.status transition happens for this attempt (checked
  235 + first thing inside trigger_org_node_full, before the real subprocess
  236 + call), so task.status is still whatever it was on entry (normally
  237 + 'queued' for a fresh dispatch) -- leaving it untouched here means the
  238 + next new_goal_dispatch tick naturally retries it, no human/Boss Agent
  239 + involvement needed. Still logged as its own execution_event (not
  240 + silently invisible) so the concurrency wait is auditable."""
  241 + if isinstance(exc, org_node_lifecycle.ConcurrencyLimitReachedError):
  242 + project_id = conn.execute("SELECT project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
  243 + conn.execute(
  244 + "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
  245 + "VALUES (?, ?, ?, 'concurrency_wait', ?)",
  246 + (str(uuid.uuid4()), project_id, task_id, json.dumps({"detail": str(exc)})),
  247 + )
  248 + conn.commit()
  249 + return
223 250 reason = scheduler._blocked_reason_for(exc) if isinstance(exc, scheduler._ROUTABLE_EXCEPTIONS) else "execution_error"
224 251 scheduler._block_task(conn, task_id, reason, detail=str(exc))
225 252  
... ...
src/test_task_execution.py
  View file @75162ac
... ... @@ -659,6 +659,34 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
659 659 # the real exception message must still be recoverable somewhere.
660 660 self.assertEqual(json.loads(payload)["detail"], "boom")
661 661  
  662 + def test_concurrency_limit_reached_does_not_mark_blocked(self):
  663 + """Real incident (2026-08-13, task 71868db6): max_concurrent_goals
  664 + backpressure got routed through the same blocked_pending_human/
  665 + execution_error bucket as a genuine failure -- wrong, since it's
  666 + normal "no free slot right now" behavior, not an error needing
  667 + human judgment. org_node_lifecycle._check_concurrency() raises
  668 + before any status transition happens, so task.status must stay
  669 + untouched (still 'queued') -- the next new_goal_dispatch tick
  670 + retries it -- while the concurrency wait itself is still logged as
  671 + a real execution_event for visibility."""
  672 + conn = schema.connect(self.db_path)
  673 + conn.execute(
  674 + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
  675 + "VALUES ('task-2', ?, 'g', 'goal', 'queued', 'planner')", (self.project_id,),
  676 + )
  677 + conn.commit()
  678 +
  679 + exc = org_node_lifecycle.ConcurrencyLimitReachedError("max_concurrent_goals=3 reached (4 active)")
  680 + task_execution._mark_blocked(conn, "task-2", exc)
  681 +
  682 + row = conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-2'").fetchone()
  683 + self.assertEqual(row, ("queued", None))
  684 + payload = conn.execute(
  685 + "SELECT payload FROM execution_event WHERE task_id = 'task-2' AND event_type = 'concurrency_wait'"
  686 + ).fetchone()[0]
  687 + self.assertIn("max_concurrent_goals=3", json.loads(payload)["detail"])
  688 + conn.close()
  689 +
662 690  
663 691 if __name__ == "__main__":
664 692 unittest.main()
... ...