宜博 / lemongraph

"""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
    ]