"""c_task_list / c_task_get (03-C §3.2): read-only task queries for the console (v_task_board, v_task_detail). Unowned by any Unit until now -- built here because U-025 is the first Unit that actually needs them. c_task_get's output goes beyond 03-C's literal "task + artifact 列表 + bug_record 列表" wording: v_task_detail (02-V §2.6) also needs the task's work_graph_run(s) (for the to-do tree built from graph_manifest, 01-M 1.3's "补充说明") and each run's work_node_execution rows (status per to-do node, expandable input/output/reflection_notes) -- 03-C's own wording undersells what the view it feeds actually requires, so this returns the fuller shape rather than forcing the frontend to make N more round trips. """ import json from src import work_graph_plan _SORT_COLUMNS = ("created_at", "updated_at", "goal_text", "status") class InvalidParamError(Exception): """INVALID_PARAM -- e.g. an unrecognized `sort` column.""" def list_tasks(conn, project_id, status=None, org_node=None, keyword=None, sort=None, order=None): """`sort`/`order` (2026-08-03, v_goal_list): defaults preserve the original `updated_at DESC` behavior every existing caller (v_task_board) already depends on. `sort` is validated against a fixed column whitelist and never interpolated as anything other than one of those literal strings -- this is the one query in this module built as a dynamic ORDER BY, so it's worth being explicit that user input never reaches the SQL text itself, only a pre-approved column name.""" query = ( "SELECT task_id, goal_text, task_type, status, assigned_org_node, loop_round, " "created_at, updated_at, execution_priority FROM task WHERE project_id = ?" ) params = [project_id] if status: query += " AND status = ?" params.append(status) if org_node: query += " AND assigned_org_node = ?" params.append(org_node) if keyword: query += " AND goal_text LIKE ?" params.append("%" + keyword + "%") sort_column = sort or "updated_at" if sort_column not in _SORT_COLUMNS: raise InvalidParamError("sort must be one of {}, got {!r}".format(_SORT_COLUMNS, sort)) sort_direction = "ASC" if (order or "desc").lower() == "asc" else "DESC" query += " ORDER BY {} {}".format(sort_column, sort_direction) rows = conn.execute(query, params).fetchall() return [ { "task_id": r[0], "goal_text": r[1], "task_type": r[2], "status": r[3], "assigned_org_node": r[4], "loop_round": r[5], "created_at": r[6], "updated_at": r[7], "execution_priority": r[8], } for r in rows ] class TaskNotFoundError(Exception): """NOT_FOUND -- 03-C §3.2: c_task_get on a nonexistent task_id.""" class TaskNotEditableError(Exception): """A goal's `goal_text` can only be edited or deleted while it's still `queued` -- once Planner/Coder/Validator has started real work from it (real artifacts, work_graph_runs, bug_records all reference task_id), changing or removing the goal out from under that history would leave those records pointing at a goal that no longer matches what was actually built, which is exactly the kind of untraceable drift this codebase otherwise goes out of its way to avoid.""" def update_task_goal_text(conn, task_id, goal_text): row = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone() if row is None: raise TaskNotFoundError(task_id) if row[0] != "queued": raise TaskNotEditableError( "task {} is status={}, goal_text can only be edited while queued".format(task_id, row[0]) ) conn.execute( "UPDATE task SET goal_text = ?, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?", (goal_text, task_id) ) conn.commit() return get_task(conn, task_id) def update_task_execution_priority(conn, task_id, execution_priority): """2026-08-13 (real incident retrospective): unlike goal_text, this is NOT gated to status='queued' -- execution_priority is a scheduling/ display hint (v_task_board's new column), not a planning input that invalidates work already done, so it should stay settable regardless of a task's current status (including a task that's already blocked or mid-flight). None clears it back to unset.""" row = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone() if row is None: raise TaskNotFoundError(task_id) if execution_priority is not None and not isinstance(execution_priority, (int, float)): raise InvalidParamError("execution_priority must be a number or null, got {!r}".format(execution_priority)) conn.execute( "UPDATE task SET execution_priority = ?, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?", (execution_priority, task_id), ) conn.commit() return get_task(conn, task_id) def delete_task(conn, task_id): row = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone() if row is None: raise TaskNotFoundError(task_id) if row[0] != "queued": raise TaskNotEditableError( "task {} is status={}, can only be deleted while queued".format(task_id, row[0]) ) # a real DELETE against a real queued task raised sqlite3.IntegrityError # (2026-08-03, found on first real verification pass) -- task_create. # create_task() always writes an execution_event row in the same # transaction as the task itself (01-M's own FK, no ON DELETE CASCADE), # so even a task that has never been touched by Planner/Coder/Validator # already has at least one execution_event referencing it. Those rows # are this task's own creation-time bookkeeping, not independent audit # content worth preserving once the task itself is gone -- a queued # task can't yet have any artifact/bug_record/work_graph_run rows # (those only start existing once real work begins), so this is the # only FK that can actually be in the way here. conn.execute("DELETE FROM execution_event WHERE task_id = ?", (task_id,)) conn.execute("DELETE FROM task WHERE task_id = ?", (task_id,)) conn.commit() def get_task(conn, task_id): row = conn.execute( "SELECT task_id, project_id, goal_text, task_type, origin_ref, created_via, status, route, " "loop_round, blocked_reason, assigned_org_node, created_at, updated_at, execution_priority " "FROM task WHERE task_id = ?", (task_id,), ).fetchone() if row is None: raise TaskNotFoundError(task_id) task = { "task_id": row[0], "project_id": row[1], "goal_text": row[2], "task_type": row[3], "origin_ref": row[4], "created_via": row[5], "status": row[6], "route": row[7], "loop_round": row[8], "blocked_reason": row[9], "assigned_org_node": row[10], "created_at": row[11], "updated_at": row[12], "execution_priority": row[13], } # 2026-08-13 (real incident): task.blocked_reason is only ever a coarse # category ('execution_error', ...) -- the actual human-readable cause # (task_execution._mark_blocked's str(exc), or org_loop_route's own # detail strings) only ever lived in the matching task_status_changed # execution_event's payload, which v_task_detail's alert banner never # surfaced -- so a blocked task showed "execution_error, needs human # intervention" with zero way to tell WHY short of querying the DB # directly. Most recent event landing on blocked_pending_human, if any. if task["status"] == "blocked_pending_human": event_row = conn.execute( "SELECT payload FROM execution_event WHERE task_id = ? AND event_type = 'task_status_changed' " "ORDER BY created_at DESC, rowid DESC LIMIT 20", (task_id,), ).fetchall() task["blocked_detail"] = None for (payload_json,) in event_row: payload = json.loads(payload_json) if payload.get("to_status") == "blocked_pending_human" and payload.get("detail"): task["blocked_detail"] = payload["detail"] break artifacts = conn.execute( "SELECT artifact_id, artifact_type, produced_by_run_id, file_path, git_commit_sha, " "validation_status, created_at FROM artifact WHERE task_id = ? ORDER BY created_at", (task_id,), ).fetchall() task["artifacts"] = [ { "artifact_id": a[0], "artifact_type": a[1], "produced_by_run_id": a[2], "file_path": a[3], "git_commit_sha": a[4], "validation_status": a[5], "created_at": a[6], } for a in artifacts ] bugs = conn.execute( "SELECT bug_id, source, description, root_cause_category, suggested_route, status, repro_steps " "FROM bug_record WHERE task_id = ?", (task_id,), ).fetchall() task["bug_records"] = [ { "bug_id": b[0], "source": b[1], "description": b[2], "root_cause_category": b[3], "suggested_route": b[4], "status": b[5], "repro_steps": b[6], } for b in bugs ] runs = conn.execute( "SELECT run_id, graph_manifest, status, started_at, completed_at, " "parent_run_id, fork_of_node_exec_id " "FROM work_graph_run WHERE task_id = ? ORDER BY started_at", (task_id,), ).fetchall() task["work_graph_runs"] = [] for w in runs: nodes = _work_nodes_for_run(conn, w[0]) graph_manifest = json.loads(w[1]) if w[1] else None task["work_graph_runs"].append({ "run_id": w[0], "graph_manifest": _display_graph_manifest(graph_manifest, nodes), "status": w[2], "started_at": w[3], "completed_at": w[4], # D (2026-08-06): both None for a normal run; both set when this # run was forked from a historical node (src/work_graph_fork.py). "parent_run_id": w[5], "fork_of_node_exec_id": w[6], # 2026-08-12 (Org Graph unification Phase 2c §2.1.1): replaces # the removed single-value work_graph_run.org_node -- a merged # run genuinely spans several org_nodes now, computed from this # run's own nodes (already fetched below for "nodes") rather # than a second query. Same shape as work_graph_run_query.py's # get_work_graph_run/list_work_graph_runs, kept in lockstep per # this module's own docstring. "org_nodes_involved": sorted({n["subagent_type"] for n in nodes if n["subagent_type"]}), "nodes": nodes, # 2026-08-08 (Work Graph self-planning + loop-back): the console's # v_graph_visualizer needs this to actually draw a loop_back's # on_failure edge / a retry_self node's "sent back Nx" badge -- # neither is derivable from graph_manifest/work_node_execution # alone (loop_back_triggered is execution_event-only by design, # see docs/done/work-graph-self-planning-and-loop-back-design.md §7). "loop_back_events": _loop_back_events_for_run(conn, w[0]), }) # 2026-08-13 (Org Graph unification Phase 3, FR-5a): planner_reviewer has # no self-planned work_graph_run (org_node_trigger.trigger_org_node() is # one flat call, see e2e_pipeline.py::_record_planner_review_event) -- # its only real audit trail is this task-scoped event log, so the Org # meta-graph reads it here instead of faking a run for that stage. task["planner_review_events"] = _planner_review_events_for_task(conn, task_id) return task def _planner_review_events_for_task(conn, task_id): rows = conn.execute( "SELECT payload, created_at FROM execution_event " "WHERE task_id = ? AND event_type = 'planner_review_verdict' ORDER BY created_at", (task_id,), ).fetchall() events = [] for payload, created_at in rows: data = json.loads(payload) events.append({ "attempt": data["attempt"], "verdict": data["verdict"], "used_existing_artifacts": data.get("used_existing_artifacts", False), "created_at": created_at, }) return events def _loop_back_events_for_run(conn, run_id): rows = conn.execute( "SELECT payload, created_at FROM execution_event " "WHERE run_id = ? AND event_type = 'loop_back_triggered' ORDER BY created_at", (run_id,), ).fetchall() events = [] for payload, created_at in rows: data = json.loads(payload) events.append({ "source_node_id": data["source_node_id"], "target_node_id": data["target_node_id"], "attempt_number": data["attempt_number"], "feedback": data.get("feedback") or [], "created_at": created_at, }) return events def _display_graph_manifest(graph_manifest, nodes): """v_task_detail's to-do tree/DAG (renderTodoTree/renderGraphVisualizerPanel in frontend/app.js) render straight off graph_manifest.nodes, in array order -- but work_graph_run.graph_manifest itself deliberately never includes the planning call that produced it (work_graph_plan.py's _record_planning_node docstring: run_graph()/fork/append-mode all read that column too and must only ever see real, executable nodes). This prepends a display-only node for that planning call -- sourced from the real work_node_execution row _record_planning_node already wrote (same node_exec_id, same log, same output_payload every other node gets) -- so it shows up as the first, clickable entry in both views without the persisted manifest itself ever changing. Runs from before this feature existed have no such row and come back unchanged.""" if graph_manifest is None: return None if not any(n["node_id"] == work_graph_plan.PLANNING_NODE_ID for n in nodes): return graph_manifest planning_node = { "node_id": work_graph_plan.PLANNING_NODE_ID, "artifact_type": "graph_manifest", "depends_on": [], } return {"nodes": [planning_node] + graph_manifest["nodes"], "edges": graph_manifest.get("edges", [])} def _work_nodes_for_run(conn, run_id): # input_schema_ref/output_schema_ref/quota_limit_type/quota_wait_started_at # added for U-026's hover tooltip + waiting_for_quota display (02-V §2.7: # "tooltip 展示 input_schema_ref/output_schema_ref/retry_count 摘要... # waiting_for_quota 态额外展示 quota_limit_type 和已等待时长") -- not # needed by U-025's to-do tree, which is why they weren't here yet. rows = conn.execute( "SELECT node_exec_id, node_id, subagent_type, status, retry_count, input_payload, " "output_payload, reflection_notes, git_worktree_ref, input_schema_ref, output_schema_ref, " "quota_limit_type, quota_wait_started_at " "FROM work_node_execution WHERE run_id = ? ORDER BY node_exec_id", (run_id,), ).fetchall() return [ { "node_exec_id": n[0], "node_id": n[1], "subagent_type": n[2], "status": n[3], "retry_count": n[4], "input_payload": json.loads(n[5]) if n[5] else None, "output_payload": json.loads(n[6]) if n[6] else None, "reflection_notes": n[7], "git_worktree_ref": n[8], "input_schema_ref": n[9], "output_schema_ref": n[10], "quota_limit_type": n[11], "quota_wait_started_at": n[12], } for n in rows ]