"""c_org_node_trigger (03-C §3.3): subprocess wrapper skeleton. U-005 scope: shell out `claude --agent -p ...` and return the text response -- "骨架，不含完整状态机" per development-plan.md. No task-lifecycle state machine yet: TASK_NOT_READY/CONCURRENCY_LIMIT_REACHED/USAGE_LIMIT_REACHED handling is U-010's job, not this one. Identity comes entirely from `.claude/agents/.md` in the target project (see src/agent_profile.py) -- this module only assembles the prompt and parses the response, no role-isolation logic of its own (03-C §3.3). **`trigger_org_node_structured` (2026-08, work-graph-node-execution- deblackbox-requirements.md)**: a sibling of `trigger_org_node` above, not a replacement -- same flat `claude --agent` call, but also extracts a JSON payload matching a caller-supplied output_schema from the free-text response (05 §5.4.1.1's documented workaround: `--json-schema` silently drops `structured_output` when combined with `--agent`, so ask for a single ```json fenced block in the prompt instead and parse it out, exactly the convention `work_graph_plan.py::plan_work_graph()` already validated in production). This is what closes the three remaining call sites that used to render a one-off Dynamic Workflow `.js` script and run it via `work_graph_execute.execute_work_graph()` purely to get "agent identity + schema-shaped output" -- `plan_work_graph()` proves flat calls already do both, so the Workflow indirection was never load-bearing for that need. """ import json import subprocess from src.agent_profile import ORG_NODES from src.work_graph_plan import extract_json_payload class OrgNodeTriggerError(Exception): """Raised when the claude subprocess exits non-zero or reports is_error.""" class StructuredOutputExtractionError(OrgNodeTriggerError): """trigger_org_node_structured got a well-formed claude response, but couldn't find any JSON payload in it matching output_schema's top-level required keys (bare or ```json-fenced) -- see extract_json_payload.""" # REQ-14's happy case cites "You've hit your session limit" as the real # signal text (session/weekly/opus phrase matching, ported verbatim from # work_graph_execute.py's own `_quota_limit_type` -- see this module's # docstring for why trigger_org_node_structured needs THIS granular version # (with a quota_limit_type of "session"/"weekly"/"opus") rather than # org_node_lifecycle.py's simpler boolean `_looks_like_usage_limit`: # work_node_run.raise_usage_limit_signal(quota_limit_type) and # _make_run_fn's own quota_limit_type-keyed logging both need the specific # type, not just a yes/no. org_node_lifecycle.py's own detection is a # different call path (trigger_org_node_full, task-lifecycle transitions, # untouched by this migration) and is deliberately left alone rather than # unified here -- merging it would touch that unrelated flow's established, # tested behavior for no benefit to this migration. work_graph_execute.py # keeps its own copy too (frozen/historical per this migration's FR-6, not # worth coupling a soon-to-be-unused module to this one) -- both copies # implement the identical algorithm, so there is no *inconsistent* duplicate # detection logic in play, only a frozen historical copy next to the live one. _USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit") 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 class UsageLimitReachedError(OrgNodeTriggerError): """USAGE_LIMIT_REACHED -- 04-P p_usage_limit_wait. `quota_limit_type` is "session"/"weekly"/"opus" (see _quota_limit_type above); callers that already have a downgrade-retry mechanism (work_node_run.py's U-020 model-downgrade retry, driven via work_node_run.raise_usage_limit_signal) consume it directly.""" def __init__(self, message, quota_limit_type): super().__init__(message) self.quota_limit_type = quota_limit_type _JSON_FENCE_INSTRUCTION = ( "Your final answer MUST be a single JSON object wrapped in exactly one " "```json fenced code block, matching this schema: {schema}. Do not " "describe what you did or are about to do -- the fenced block's content " "IS the answer, not a summary of it." ) def trigger_org_node_structured(org_node, prompt, target_folder_path, output_schema, model=None, timeout=300): """Like trigger_org_node(), but also extracts a JSON payload matching output_schema's top-level required keys from the free-text response -- see this module's docstring. `_JSON_FENCE_INSTRUCTION` is appended to `prompt` automatically, callers don't repeat this boilerplate themselves. `model`: real CLI `--model` override (FR-4) -- empirically verified for real (RUN_REAL_AGENT_E2E=1, see this module's own test file) that `claude --agent -p ... --model ` (1) still loads the target org_node's persona, (2) genuinely runs the requested model (the JSON response's own `modelUsage` key names it), and (3) the ```json fence-extraction convention above still works on that real output -- 05 §5.4.1.5's "CLI top-level args and DSL-internal options are two different paths, don't assume one from the other" lesson, checked here rather than assumed. """ if org_node not in ORG_NODES: raise ValueError(org_node) args = [ "claude", "--agent", org_node, "-p", prompt + "\n\n" + _JSON_FENCE_INSTRUCTION.format(schema=json.dumps(output_schema)), ] if model: args += ["--model", model] args += ["--output-format", "json", "--permission-mode", "auto"] result = subprocess.run( args, cwd=target_folder_path, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=timeout, ) if result.returncode != 0: 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( "trigger_org_node_structured for {} exited {}\n{}".format(org_node, result.returncode, detail), quota_type, ) raise OrgNodeTriggerError( "trigger_org_node_structured for {} exited {}\n{}".format(org_node, result.returncode, detail) ) payload = json.loads(result.stdout) if payload.get("is_error"): quota_type = _quota_limit_type(str(payload.get("result", ""))) if quota_type: raise UsageLimitReachedError( "trigger_org_node_structured for {} hit a usage limit: {}".format(org_node, payload), quota_type ) raise OrgNodeTriggerError( "trigger_org_node_structured for {} reported is_error: {}".format(org_node, payload) ) required_keys = list(output_schema.get("required", [])) extracted = extract_json_payload(payload.get("result", ""), required_keys=required_keys) if extracted is None: # 2026-08-15 (real incident, task b243dfe8): this used to truncate # to [:2000] chars -- fine for the "process exited nonzero" error # paths above, where a short excerpt is enough context, but actively # counterproductive here: the WHOLE POINT of this exception is # diagnosing why extraction failed (genuine model-side truncation # vs some other JSON-validity issue), and a truncated preview can # LOOK like the response was cut off mid-generation when that's # really just this display limit -- indistinguishable from the # real thing without the full text, which is exactly what happened # investigating this live. No length cap: this flows into # task_execution._mark_blocked's str(exc), stored as a plain TEXT # execution_event column with no practical size limit, so keeping # the complete response is the honest, actually-diagnosable choice. raise StructuredOutputExtractionError( "could not extract a payload matching required keys {} from org_node={}'s response: {!r}".format( required_keys, org_node, payload.get("result", "") ) ) return extracted def trigger_org_node(org_node, prompt, target_folder_path, timeout=300): """Runs `claude --agent -p ` with cwd=target_folder_path. `.claude/agents/.md` must already be deployed in target_folder_path (agent_profile.deploy_agent_definitions) -- this function does not deploy it. Returns the response text (the JSON `result` field). """ if org_node not in ORG_NODES: raise ValueError(org_node) result = subprocess.run( [ "claude", "--agent", org_node, "-p", prompt, "--output-format", "json", "--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: a real Coder call once failed with returncode=1 and # empty stderr, and the exception message (stderr-only, as it used to # be) gave zero diagnostic signal -- whatever claude actually said # about the failure was in stdout, which this branch was discarding. # Both are now included (truncated -- stdout in JSON mode can be # large) so a future real failure is actually debuggable. raise OrgNodeTriggerError( "claude --agent {} exited {}\nstderr: {}\nstdout (truncated): {}".format( org_node, result.returncode, result.stderr, result.stdout[:4000] ) ) payload = json.loads(result.stdout) if payload.get("is_error"): raise OrgNodeTriggerError("claude --agent {} reported is_error: {}".format(org_node, payload)) return payload["result"]