"""c_org_node_trigger (03-C §3.3): full implementation.

U-010 scope: per 05 §5.1 4.4's staged rollout (05-实施约束与依赖.md §5.5:
"手工调用验证 Planner/Coder/Validator 三个 Org Agent 分别可被触发" -- NOT
architecture doc §11, which doesn't contain that phrase; development-plan.md's
citation was wrong and is corrected in this session's notes), this drives the
three DIRECT forward transitions in `p_org_task_lifecycle` (04-P) that a single
`c_org_node_trigger` call causes:

    queued -> planning_brd_prd_sdd   (trigger planner)
    planning_brd_prd_sdd -> coding   (trigger coder)
    coding -> validating             (trigger validator)

with TASK_NOT_READY / CONCURRENCY_LIMIT_REACHED / USAGE_LIMIT_REACHED guards.

Everything past that -- planning_failed/coding_failed retries, the
validator_pass/validator_fail decision and bug_record-driven routing,
blocked_pending_human -- needs real Work Graph artifact production (U-011+)
or Org-level Loop routing (U-015/U-016, PH2) that doesn't exist yet, and is
not attempted here. The artifact-validation guards below are real, testable
logic even though nothing produces passing artifacts for real yet -- tests
seed the precondition rows directly, same as any consumer of an
upstream contract that hasn't been built.
"""

import json
import uuid

from src.org_node_trigger import OrgNodeTriggerError, trigger_org_node

_PRE_STATE = {
    "planner": "queued",
    "planner_reviewer": "planning_brd_prd_sdd",
    "coder": "reviewing_planner_output",
    "validator": "coding",
}
_POST_STATE = {
    "planner": "planning_brd_prd_sdd",
    "planner_reviewer": "reviewing_planner_output",
    "coder": "coding",
    "validator": "validating",
}
# Org Graph unification Phase 1 (2026-08-08): _REQUIRED_ARTIFACTS["coder"] used to be
# ("brd", "prd", "sdd", "benchmark"), but artifact_write.py sets validation_status='passed'
# unconditionally on write (no real semantic check exists yet -- see that module's own
# docstring), so this gate never actually blocked anything. planner_reviewer is now the
# first real semantic gate before Coder starts (see e2e_pipeline.review_planner_output).
# Only `sdd` is kept here, as pure existence-defense: implement_code() requires a real
# sdd_content positional argument, and a missing sdd artifact should fail clearly here
# rather than crash deeper inside implement_code() with a harder-to-diagnose error.
_REQUIRED_ARTIFACTS = {
    "coder": ("sdd",),
    "validator": ("code_diff", "test_report"),
}
# REQ-14 happy case cites "You've hit your session limit" as the real signal
# text; weekly/generic phrasing added defensively. Best-effort until we've
# observed the real error text from an actual exhausted-quota call (U-020).
_USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit")


class TaskNotReadyError(Exception):
    """TASK_NOT_READY -- 03-C §3.3."""


class ConcurrencyLimitReachedError(Exception):
    """CONCURRENCY_LIMIT_REACHED -- 03-C §3.3."""


class UsageLimitReachedError(Exception):
    """USAGE_LIMIT_REACHED -- 03-C §3.3 / 04-P p_usage_limit_wait. Detection
    only here; the wait-and-retry loop itself is U-012/U-020's job."""


def _get_task(conn, task_id):
    row = conn.execute("SELECT status, project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()
    if row is None:
        raise TaskNotReadyError("no such task: " + task_id)
    return {"status": row[0], "project_id": row[1]}


def _artifacts_ready(conn, task_id, artifact_types):
    for artifact_type in artifact_types:
        row = conn.execute(
            "SELECT 1 FROM artifact WHERE task_id = ? AND artifact_type = ? AND validation_status = 'passed'",
            (task_id, artifact_type),
        ).fetchone()
        if row is None:
            return False
    return True


def _check_concurrency(conn):
    row = conn.execute(
        "SELECT max_concurrent_goals FROM org_concurrency_config WHERE config_id = 'default'"
    ).fetchone()
    limit = row[0] if row else 1  # 01-M 1.11 default
    active = conn.execute(
        "SELECT COUNT(*) FROM task WHERE status NOT IN ('queued', 'done', 'blocked_pending_human')"
    ).fetchone()[0]
    if active >= limit:
        raise ConcurrencyLimitReachedError(
            "max_concurrent_goals={} reached ({} active)".format(limit, active)
        )


def _looks_like_usage_limit(message):
    lowered = message.lower()
    return any(phrase in lowered for phrase in _USAGE_LIMIT_PHRASES)


def trigger_org_node_full(conn, org_node, task_id, target_folder_path, prompt, timeout=300):
    """Full c_org_node_trigger: guards + real subprocess call (src.org_node_trigger)
    + task.status transition, all in one call. Returns the updated
    {"task_id", "status", "assigned_org_node"}.
    """
    if org_node not in _PRE_STATE:
        raise ValueError(org_node)

    task = _get_task(conn, task_id)
    if task["status"] != _PRE_STATE[org_node]:
        raise TaskNotReadyError(
            "task {} is in status={}, not ready for {} (needs {})".format(
                task_id, task["status"], org_node, _PRE_STATE[org_node]
            )
        )
    if org_node in _REQUIRED_ARTIFACTS and not _artifacts_ready(conn, task_id, _REQUIRED_ARTIFACTS[org_node]):
        raise TaskNotReadyError("required artifacts not validation_status=passed yet for task " + task_id)
    if org_node == "planner":
        _check_concurrency(conn)

    try:
        trigger_org_node(org_node, prompt, target_folder_path, timeout=timeout)
    except OrgNodeTriggerError as e:
        if _looks_like_usage_limit(str(e)):
            raise UsageLimitReachedError(str(e)) from e
        raise

    old_status = task["status"]
    new_status = _POST_STATE[org_node]
    conn.execute(
        "UPDATE task SET status = ?, assigned_org_node = ?, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
        (new_status, org_node, task_id),
    )
    conn.execute(
        "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
        "VALUES (?, ?, ?, 'task_status_changed', ?)",
        (
            str(uuid.uuid4()),
            task["project_id"],
            task_id,
            json.dumps({"from_status": old_status, "to_status": new_status}),
        ),
    )
    conn.commit()
    return {"task_id": task_id, "status": new_status, "assigned_org_node": org_node}