"""c_work_graph_execute (03-C §3.5): dynamic-workflow invocation, full implementation.

**HISTORICAL, 2026-08 (work-graph-node-execution-deblackbox-requirements.md)**:
this module's `write_workflow_script`/`execute_work_graph`/`execute_work_
graph_with_usage_limit_retry` are no longer called by any production code
path -- the three call sites that used to render a one-off Dynamic Workflow
`.js` script here purely to get "agent identity + schema-shaped output" now
use `org_node_trigger.trigger_org_node_structured()` (a flat `claude
--agent` call + ```json-fence extraction), the same mechanism `work_graph_
plan.py::plan_work_graph()` had already been using in production the whole
time. Kept in this file (not deleted) as a historical record and because
`.claude/workflows/*.js` files already generated by real self-hosting runs
are real execution artifacts worth being able to re-read the generating
code for. Do not add new callers -- extend `org_node_trigger.py` instead.
See that module's docstring for the full rationale.

Writes an orchestration script into the target project's `.claude/workflows/
<name>.js` and invokes it via `claude -p 'Run /<name> with args {...}'`,
reusing the exact code path the A-1 spike validated (spikes/a1-dynamic-workflows/,
confirmed 2026-07-29).

U-012 scope, added on top of U-006's skeleton: CLASSIFIER_BLOCKED detection
(03-C §3.5) and USAGE_LIMIT_REACHED backoff-retry (04-P §4.4.1
p_usage_limit_wait -- REQ-14's happy case only; boundary/failure is U-020).
Both are pure, time-injectable logic wrapping the same subprocess call, no
dependency on U-013's work_node_execution rows (which don't exist yet --
see development-plan.md's 2026-07-31 scope note on this Unit for why the
retry loop operates at the whole-work-graph level rather than per node).
"""

import json
import os
import subprocess
import time

_USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit")
_BACKOFF_SCHEDULE_MINUTES = (5, 10, 20, 30)  # then repeats at 30 until the type's timeout
_TIMEOUT_SECONDS = {"session": 8 * 3600, "weekly": 9 * 24 * 3600}
# 03-C §3.5: "分类器连续拦截 3 次或累计 20 次" -- a single `-p` call's own
# permission_denials array already reflects that whole session's denials, so
# 3 is used as the practical per-call threshold (conservative relative to the
# 20 cumulative ceiling; true cross-call cumulative tracking needs U-013's
# per-node persistence and isn't attempted here).
_CLASSIFIER_BLOCKED_THRESHOLD = 3


class WorkGraphExecuteError(Exception):
    """Raised when the claude subprocess exits non-zero, reports is_error, or
    produces no structured_output, and it isn't one of the more specific
    errors below."""


class ClassifierBlockedError(WorkGraphExecuteError):
    """CLASSIFIER_BLOCKED -- 03-C §3.5. Not retryable; routes to
    blocked_pending_human, blocked_reason='classifier_blocked' (04-P)."""


class UsageLimitReachedError(WorkGraphExecuteError):
    """USAGE_LIMIT_REACHED -- 04-P p_usage_limit_wait. Retryable with backoff;
    see execute_work_graph_with_usage_limit_retry."""

    def __init__(self, message, quota_limit_type):
        super().__init__(message)
        self.quota_limit_type = quota_limit_type


class UsageLimitWaitTimeoutError(WorkGraphExecuteError):
    """p_usage_limit_wait exceeded its timeout -> blocked_pending_human,
    blocked_reason='usage_limit_wait_timeout' (04-P §4.4.1)."""


def _quota_limit_type(text):
    lowered = text.lower()
    if "weekly" in lowered and "hit your" in lowered:
        return "weekly"
    if "opus" in lowered and ("limit" in lowered or "quota" in lowered):
        return "opus"
    if any(phrase in lowered for phrase in _USAGE_LIMIT_PHRASES):
        return "session"
    return None


def write_workflow_script(target_folder_path, name, script_content):
    """Writes `<target>/.claude/workflows/<name>.js` -- versioned with the
    target project's own git history (03-C §3.5 step 1), not graphlemon's.

    Historical mechanism, no longer called by production code as of 2026-08
    (see this module's docstring) -- do not add new callers."""
    workflows_dir = os.path.join(target_folder_path, ".claude", "workflows")
    os.makedirs(workflows_dir, exist_ok=True)
    path = os.path.join(workflows_dir, name + ".js")
    with open(path, "w", encoding="utf-8") as f:
        f.write(script_content)
    return path


def execute_work_graph(target_folder_path, name, args, output_schema, timeout=600):
    """Runs the workflow already written to `.claude/workflows/<name>.js`
    (write_workflow_script) via `claude -p 'Run /<name> with args {...}'`.

    No `--bare` (A-1 spike: skips credential/login discovery, unrelated to
    workflow presence). `--permission-mode auto`, never `bypassPermissions`/
    `--dangerously-skip-permissions` (05 §5.3.1). Returns the parsed
    `structured_output` dict. Raises ClassifierBlockedError or
    UsageLimitReachedError instead of the generic WorkGraphExecuteError when
    the response matches those specific signals.

    Historical mechanism, no longer called by production code as of 2026-08
    (see this module's docstring) -- do not add new callers.
    """
    prompt = "Run /{} with args {}".format(name, json.dumps(args))
    result = subprocess.run(
        [
            "claude", "-p", prompt,
            "--output-format", "json",
            "--json-schema", json.dumps(output_schema),
            "--permission-mode", "auto",
        ],
        cwd=target_folder_path,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if result.returncode != 0:
        # 2026-07-31 fix (same bug as org_node_trigger.py): stdout can carry
        # the actual error detail even on nonzero exit, and messages here
        # were dropping it entirely -- check/include both streams.
        quota_type = _quota_limit_type(result.stderr) or _quota_limit_type(result.stdout)
        detail = "stderr: {}\nstdout (truncated): {}".format(result.stderr, result.stdout[:4000])
        if quota_type:
            raise UsageLimitReachedError(
                "workflow {} exited {}\n{}".format(name, result.returncode, detail), quota_type
            )
        raise WorkGraphExecuteError("workflow {} exited {}\n{}".format(name, result.returncode, detail))

    payload = json.loads(result.stdout)
    if payload.get("is_error"):
        denials = payload.get("permission_denials") or []
        if len(denials) >= _CLASSIFIER_BLOCKED_THRESHOLD:
            raise ClassifierBlockedError(
                "workflow {} blocked by permission classifier ({} denials): {}".format(
                    name, len(denials), payload
                )
            )
        quota_type = _quota_limit_type(str(payload.get("result", "")))
        if quota_type:
            raise UsageLimitReachedError("workflow {} hit a usage limit: {}".format(name, payload), quota_type)
        raise WorkGraphExecuteError("workflow {} reported is_error: {}".format(name, payload))

    structured_output = payload.get("structured_output")
    if structured_output is None:
        raise WorkGraphExecuteError("workflow {} produced no structured_output: {}".format(name, payload))
    return structured_output


def execute_work_graph_with_usage_limit_retry(
    target_folder_path, name, args, output_schema, timeout=600,
    sleep=time.sleep, now=time.monotonic,
):
    """Wraps execute_work_graph with p_usage_limit_wait (04-P §4.4.1): a
    session/weekly quota hit backs off 5->10->20->30min (repeating at 30)
    and retries until that type's timeout, then raises
    UsageLimitWaitTimeoutError. ClassifierBlockedError is never retried
    (03-C: not retryable, needs human review) and propagates immediately.

    An opus quota hit skips backoff and raises immediately, unlike
    src/work_node_run.py's run_work_node (U-020), which actually retries on
    a downgraded model. Deliberately not matched here: REQ-14's coverage
    (06 节) only names c_work_node_run/c_org_node_trigger, not
    c_work_graph_execute, and this function's underlying mechanism (a
    dynamic-workflow script with no single top-level --agent identity, per
    03-C §3.5) has no one "the model" to swap the way a single-agent call
    does -- individual agent() calls inside the workflow script would each
    need their own override, a deeper A-1 dynamic-workflow question this
    Unit doesn't attempt. Caller still has to decide what "retry on a
    non-opus model" means here.

    `sleep`/`now` are injectable so tests don't block on real 5-30 minute waits.

    Historical mechanism, no longer called by production code as of 2026-08
    (see this module's docstring) -- do not add new callers. FR-5's
    equivalent retry-on-usage-limit behavior for the flat-call mechanism is
    `work_node_run.py`'s own model-downgrade retry (U-020), driven via
    `org_node_trigger.UsageLimitReachedError`/`work_node_run.
    raise_usage_limit_signal` -- not a direct port of this function.
    """
    start = now()
    backoff_index = 0
    while True:
        try:
            return execute_work_graph(target_folder_path, name, args, output_schema, timeout=timeout)
        except UsageLimitReachedError as e:
            if e.quota_limit_type == "opus":
                raise UsageLimitWaitTimeoutError(
                    "opus quota exhausted -- caller should retry on a non-opus model: " + str(e)
                ) from e
            elapsed = now() - start
            timeout_seconds = _TIMEOUT_SECONDS[e.quota_limit_type]
            if elapsed >= timeout_seconds:
                raise UsageLimitWaitTimeoutError(
                    "{} quota wait exceeded {}s timeout after {:.0f}s: {}".format(
                        e.quota_limit_type, timeout_seconds, elapsed, e
                    )
                ) from e
            wait_minutes = _BACKOFF_SCHEDULE_MINUTES[min(backoff_index, len(_BACKOFF_SCHEDULE_MINUTES) - 1)]
            sleep(wait_minutes * 60)
            backoff_index += 1