org_node_trigger.py
9.09 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
"""c_org_node_trigger (03-C §3.3): subprocess wrapper skeleton.
U-005 scope: shell out `claude --agent <org_node> -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/<org_node>.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 <name> -p ... --model <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:
raise StructuredOutputExtractionError(
"could not extract a payload matching required keys {} from org_node={}'s response: {!r}".format(
required_keys, org_node, payload.get("result", "")[:2000]
)
)
return extracted
def trigger_org_node(org_node, prompt, target_folder_path, timeout=300):
"""Runs `claude --agent <org_node> -p <prompt>` with cwd=target_folder_path.
`.claude/agents/<org_node>.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"]