宜博 / 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

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