"""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 _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 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], } 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 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 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], } 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, org_node, 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"] = [ { "run_id": w[0], "org_node": w[1], "graph_manifest": json.loads(w[2]) if w[2] else None, "status": w[3], "started_at": w[4], "completed_at": w[5], # 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[6], "fork_of_node_exec_id": w[7], "nodes": _work_nodes_for_run(conn, w[0]), # 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/todo/work-graph-self-planning-and-loop-back-design.md §7). "loop_back_events": _loop_back_events_for_run(conn, w[0]), } for w in runs ] return task 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 _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 ]