File size: 27,064 Bytes
35676b4
 
2401ce5
7880373
35676b4
 
 
 
 
 
 
 
 
 
 
 
e6496c0
35676b4
559c2ff
7880373
b419b7b
7880373
 
 
 
 
 
35676b4
 
 
 
 
 
 
 
 
7880373
35676b4
2401ce5
7880373
 
 
 
 
 
 
2401ce5
 
 
 
 
 
 
 
 
 
 
 
 
 
7880373
 
 
 
 
 
 
2401ce5
 
35676b4
 
 
 
 
 
 
 
 
 
 
 
559c2ff
8727e56
e6496c0
35676b4
 
 
7880373
 
35676b4
 
 
 
 
 
 
7880373
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
35676b4
 
 
7880373
35676b4
7880373
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
35676b4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7880373
35676b4
 
 
b002600
 
 
 
 
 
35676b4
b002600
 
 
 
 
 
 
35676b4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7880373
 
 
 
 
 
 
 
 
 
 
b419b7b
 
 
 
 
 
 
 
 
 
 
 
7880373
 
 
 
 
 
 
 
 
b419b7b
7880373
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8727e56
7880373
 
 
559c2ff
 
 
 
 
 
 
 
 
 
e6496c0
 
 
 
559c2ff
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e6496c0
 
 
602548d
 
 
 
 
 
 
 
 
 
e6496c0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
559c2ff
 
 
e6496c0
559c2ff
 
e6496c0
 
 
 
 
 
 
559c2ff
 
e6496c0
 
559c2ff
 
 
 
 
 
 
8727e56
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7880373
 
 
35676b4
 
 
7880373
2401ce5
35676b4
 
 
2401ce5
 
 
 
 
 
 
35676b4
7880373
 
 
 
 
 
 
 
 
 
35676b4
 
602548d
7880373
 
e6496c0
7880373
 
2401ce5
567e993
 
 
 
 
 
 
 
 
7880373
 
 
 
 
567e993
 
 
 
35676b4
 
 
 
2401ce5
602548d
2401ce5
602548d
 
 
 
 
 
2401ce5
 
 
602548d
 
 
 
 
2401ce5
35676b4
 
8727e56
35676b4
7880373
 
 
559c2ff
43f40b2
7880373
 
 
b419b7b
 
 
7880373
b419b7b
 
 
7880373
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8727e56
 
 
 
 
 
7880373
 
 
 
 
 
 
35676b4
 
 
602548d
 
 
 
 
7880373
 
 
 
 
 
 
35676b4
 
559c2ff
8727e56
35676b4
 
 
 
7880373
559c2ff
8727e56
 
35676b4
 
 
7880373
35676b4
 
 
 
7880373
35676b4
 
 
7880373
 
35676b4
 
 
 
 
559c2ff
8727e56
e6496c0
35676b4
 
 
7880373
 
35676b4
 
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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
import json
import re
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

from agent.tools import (
    get_financial_metrics,
    search_filing,
    search_transcript,
    search_news,
    get_analyst_expectations,
)
from agent.prompts import SYSTEM_PROMPT, SYNTHESIS_STRUCTURED_PROMPT, language_directive
from agent.schemas import BriefOutput
from agent.post_synthesis import apply_reliability, attach_edge_signals
from agent.evidence import (
    NO_VERIFIED_SYNTHESIS_MESSAGE,
    evidence_records_from,
    evidence_envelope,
    is_usable_evidence_payload,
    parse_evidence_envelope,
)
from agent.llm import RunConfig, default_config, make_chat_model, build_system_message, ANTHROPIC_DEFAULT_MODEL

TOOLS = [
    get_financial_metrics,
    search_filing,
    search_transcript,
    search_news,
    get_analyst_expectations,
]
MAX_TOOL_ROUNDS = 10
MODEL = ANTHROPIC_DEFAULT_MODEL  # legacy alias — prefer agent.llm.RunConfig for new code

TOOL_BY_NAME = {t.name: t for t in TOOLS}
_EXPECTED_TOOL_SOURCES = {
    "get_financial_metrics": {"metrics"},
    "search_filing": {"10-K", "10-Q"},
    "search_transcript": {"transcript"},
    "search_news": {"news"},
    "get_analyst_expectations": {"analyst"},
}


def _invoke_one(tc: dict) -> ToolMessage:
    """Invoke a single tool call and return a ToolMessage."""
    name = tc["name"]
    args = tc.get("args") or {}
    tool_call_id = tc.get("id", "")
    try:
        fn = TOOL_BY_NAME.get(name)
        if fn is None:
            raise ValueError(f"Unknown tool: {name!r}")
        result = fn.invoke(args)
        content = result if isinstance(result, str) else str(result)
    except Exception as exc:
        content = evidence_envelope(
            tool=name,
            query=args,
            status="ERROR",
            message=str(exc),
            error_code="TOOL_INVOCATION_ERROR",
        )
    return ToolMessage(tool_call_id=tool_call_id, name=name, content=content)

# Tool names that the agent must touch before synthesis can proceed.
# Without filing AND transcript evidence, the structured brief schema
# (mda_summary, risks_categorized, management_commentary, etc.) cannot
# be filled without fabrication.
REQUIRED_TOOL_NAMES = {"search_filing", "search_transcript"}


class AgentState(TypedDict):
    ticker: str
    messages: Annotated[list[BaseMessage], add_messages]
    tool_round_count: int
    nudge_fired: bool
    edge_signals: Optional[list[dict]]  # precomputed deterministic signals
    profile_payloads: Optional[list[str]]
    language: Optional[str]             # e.g. "French" — prose fields in brief will use this language
    brief: Optional[dict]
    brief_markdown: Optional[str]
    synthesis_error: Optional[str]
    coverage: Optional[dict]
    verification_report: Optional[dict]


def _called_tool_names(messages: list[BaseMessage]) -> set[str]:
    return {m.name for m in messages if isinstance(m, ToolMessage) and getattr(m, "name", None)}


def _filing_call_count(messages: list[BaseMessage]) -> int:
    return sum(
        1 for m in messages
        if isinstance(m, ToolMessage)
        and getattr(m, "name", None) == "search_filing"
        and is_usable_evidence_payload(m)
    )


def _tool_envelopes(messages: list[BaseMessage], name: str) -> list[dict]:
    envelopes = []
    for message in messages:
        if not isinstance(message, ToolMessage) or getattr(message, "name", None) != name:
            continue
        envelope = parse_evidence_envelope(message)
        if envelope and envelope.get("tool") == name:
            envelopes.append(envelope)
    return envelopes


def _records_for_tool(envelope: dict, name: str):
    allowed = _EXPECTED_TOOL_SOURCES.get(name, set())
    return [
        record for record in evidence_records_from(envelope)
        if record.ref.source in allowed
    ]


def _coverage_report(messages: list[BaseMessage]) -> dict:
    groups = {
        "metrics": "get_financial_metrics",
        "filings": "search_filing",
        "transcripts": "search_transcript",
        "analyst": "get_analyst_expectations",
        "news": "search_news",
    }
    report: dict = {}
    for label, name in groups.items():
        envelopes = _tool_envelopes(messages, name)
        valid_counts = {id(e): len(_records_for_tool(e, name)) for e in envelopes}
        ok = [
            e for e in envelopes
            if e.get("status") == "OK" and valid_counts[id(e)] > 0
        ]
        invalid = [
            e for e in envelopes
            if e.get("status") == "OK" and valid_counts[id(e)] == 0
        ]
        if ok:
            status = "OK"
        elif any(e.get("status") == "ERROR" for e in envelopes):
            status = "ERROR"
        elif invalid:
            status = "INVALID"
        elif any(e.get("status") == "EMPTY" for e in envelopes):
            status = "EMPTY"
        else:
            status = "NOT_CALLED"
        report[label] = {
            "status": status,
            "successful_calls": len(ok),
            "evidence_count": sum(valid_counts[id(e)] for e in ok),
            "errors": [
                (e.get("error") or {}).get("message", "tool error")
                for e in envelopes if e.get("status") == "ERROR"
            ] + (["invalid evidence record"] if invalid else []),
        }
    gaps = []
    if report["metrics"]["evidence_count"] < 1:
        gaps.append("financial metrics unavailable")
    if report["filings"]["evidence_count"] < 1:
        gaps.append("filing evidence unavailable")
    elif report["filings"]["successful_calls"] < 2:
        gaps.append("only one filing search completed")
    if report["transcripts"]["evidence_count"] < 1:
        gaps.append("transcript evidence unavailable")
    report["gaps"] = gaps
    report["status"] = "COMPLETE" if not gaps else "PARTIAL"
    return report


def _can_synthesize(messages: list[BaseMessage]) -> bool:
    coverage = _coverage_report(messages)
    return (
        coverage["metrics"]["evidence_count"] >= 1
        and coverage["filings"]["evidence_count"] >= 1
    )


def _coverage_gaps(messages: list[BaseMessage]) -> list[str]:
    """Return retryable gaps; EMPTY is confirmed absence, not retryable."""
    gaps = []
    metrics = _tool_envelopes(messages, "get_financial_metrics")
    filings = _tool_envelopes(messages, "search_filing")
    transcripts = _tool_envelopes(messages, "search_transcript")
    if not any(e.get("status") == "OK" and _records_for_tool(e, "get_financial_metrics") for e in metrics):
        gaps.append("get_financial_metrics")
    filing_ok = sum(
        e.get("status") == "OK" and bool(_records_for_tool(e, "search_filing"))
        for e in filings
    )
    if filing_ok < 2 and not any(e.get("status") == "EMPTY" for e in filings):
        gaps.append(f"search_filing (need 2 successful calls; {filing_ok} so far)")
    if not any(
        e.get("status") == "EMPTY"
        or (e.get("status") == "OK" and _records_for_tool(e, "search_transcript"))
        for e in transcripts
    ):
        gaps.append("search_transcript")
    return gaps


def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    under_cap = state["tool_round_count"] < MAX_TOOL_ROUNDS

    if getattr(last, "tool_calls", None) and under_cap:
        return "tools"

    # Agent stopped emitting tool calls. Check coverage floor before
    # routing to synthesis: require ≥2 filing calls + ≥1 transcript call.
    if not getattr(last, "tool_calls", None) and under_cap and not state.get("nudge_fired"):
        if _coverage_gaps(state["messages"]):
            return "nudge"

    return "synthesis" if _can_synthesize(state["messages"]) else "partial"


def _extract_json(raw: str) -> str:
    """Extract the first complete JSON object from LLM output.

    Uses raw_decode so trailing content — extra commentary, a duplicate object
    (e.g. when the model was asked about multiple tickers), or a stray code
    fence — is ignored instead of triggering a json.loads 'Extra data' error.
    """
    start = raw.find("{")
    if start != -1:
        try:
            _obj, end = json.JSONDecoder().raw_decode(raw, start)
            return raw[start:end]
        except json.JSONDecodeError:
            pass
    # Last-resort fallback: original first-{ to last-} heuristic
    end = raw.rfind("}")
    if start != -1 and end != -1 and end > start:
        return raw[start:end + 1]
    return raw


def nudge_node(state: AgentState) -> dict:
    """One-shot prod when the agent tries to terminate without minimum coverage."""
    gaps = _coverage_gaps(state["messages"])
    gaps_str = " and ".join(f"`{g}`" for g in gaps)
    nudge = HumanMessage(content=(
        f"Before synthesizing, you must satisfy these coverage requirements: {gaps_str}. "
        "Filings (10-Q/10-K) are the primary source — `what_changed`, `bull_points`, "
        "`bear_points`, and `risks_categorized` must be predominantly filing-sourced. "
        "Issue the missing search call(s) now (batched where independent)."
    ))
    return {"messages": [nudge], "nudge_fired": True}


def _partial_brief(state: AgentState, reason: str) -> dict:
    coverage = _coverage_report(state.get("messages", []))
    company_name = state["ticker"].upper()
    filing_date = ""
    for envelope in _tool_envelopes(state.get("messages", []), "get_financial_metrics"):
        for record in envelope.get("records") or []:
            metadata = record.get("metadata") or {}
            company_name = metadata.get("company_name") or company_name
            filing_date = metadata.get("filing_date") or filing_date
            if filing_date:
                break
    if not filing_date or company_name == state["ticker"].upper():
        try:
            from storage.metrics_db import get_metrics

            metrics = get_metrics(state["ticker"])
            if isinstance(metrics, dict):
                if company_name == state["ticker"].upper():
                    company_name = metrics.get("company_name") or company_name
                if not filing_date:
                    filing_date = metrics.get("filing_date") or filing_date
        except Exception:
            pass
    generated_at = datetime.now(timezone.utc).isoformat()
    return {
        "ticker": state["ticker"].upper(),
        "company_name": company_name,
        "filing_date": filing_date,
        "status": "PARTIAL",
        "schema_version": "brief.v2",
        "generated_at": generated_at,
        "data_as_of": filing_date,
        "what_matters_most": NO_VERIFIED_SYNTHESIS_MESSAGE,
        "non_obvious_takeaway": "",
        "analytical_tensions": [],
        "between_the_lines": [],
        "earnings_quality_signals": [],
        "standout_number": None,
        "what_changed": [],
        "bull_points": [],
        "bear_points": [],
        "what_to_watch": [],
        "trends": [],
        "evidence_notes": [reason] + list(coverage.get("gaps") or []),
        "evidence_coverage": {
            "status": "INCOMPLETE", "verified": 0, "unverified": 0,
            "failed": 0, "total": 0,
        },
        "verification_report": {
            "verified": 0, "unverified": 0, "failed": 0, "removed": 0,
        },
        "coverage": coverage,
        "mda_summary": {
            "drivers": [], "headwinds": [], "language_shift": "", "key_quote": None,
        },
        "risks_categorized": [],
        "management_commentary": [],
        "guidance_history": [],
        "sentiment": None,
        "market_expectations": None,
        "quarter_deltas": state.get("edge_signals") or [],
        "display_policy": {
            "event_returns_aligned": False,
            "market_expectations_aligned": False,
            "aggregate_reliability_meaningful": False,
        },
        "language": state.get("language") or "English",
        "company_profile": None,
    }


def _format_signals_message(signals: list[dict]) -> str:
    """Format precomputed edge signals as a compact labelled block for the agent."""
    lines = ["== PRECOMPUTED EDGE SIGNALS (deterministic, no LLM) ==\n"]
    kind_labels = {
        "risk_added":              "NEW RISK",
        "risk_removed":            "REMOVED RISK",
        "risk_reworded":           "REWORDED RISK",
        "guidance_language_shift": "GUIDANCE LANGUAGE SHIFT",
        "term_frequency":          "TERM FREQUENCY SHIFT",
        "kpi_dropped":             "DROPPED KPI",
        "tone_trend":              "MANAGEMENT TONE TREND",
        "topic_arc":               "TRANSCRIPT TOPIC ARC",
        "recurring_evasion":       "RECURRING Q&A EVASION",
        "topic_fade":              "PREPARED-REMARKS TOPIC FADE",
    }
    for i, s in enumerate(signals, 1):
        kind = s.get("kind", "")
        label = kind_labels.get(kind, kind.upper())
        sig = s.get("significance", "MEDIUM")
        term = s.get("term", "")
        term_str = f" — {term}" if term else ""
        lines.append(f"[SIG-{i}] {label}{term_str} [{sig}]")
        if s.get("before_text"):
            lines.append(f"  BEFORE ({s.get('period_from','')}): \"{s['before_text']}\"")
        if s.get("after_text"):
            lines.append(f"  AFTER  ({s.get('period_to','')}): \"{s['after_text']}\"")
        if s.get("computed_metric"):
            lines.append(f"  METRIC: {s['computed_metric']}")
        lines.append("")
    lines.append("== END PRECOMPUTED EDGE SIGNALS ==")
    return "\n".join(lines)


MAX_EDGE_SIGNALS = 12
MAX_FILING_SIGNALS = 8
MAX_TRANSCRIPT_SIGNALS = 6
# Output budget for the single-call synthesis (brief + company profile +
# one full evidence_ref per fact). 16384 was inherited from the pre-merge
# two-call split and truncated real AAPL briefs mid-JSON (stop_reason=
# max_tokens at exactly 16384 output tokens). 64000 is the claude-haiku-4-5
# output ceiling; billing only covers tokens actually generated.
SYNTHESIS_MAX_TOKENS = 64000


class SynthesisTruncatedError(RuntimeError):
    """The synthesis stream was cut off by the max_tokens limit."""


def _cap_signals(signals: list[dict], max_total: int = MAX_EDGE_SIGNALS) -> list[dict]:
    """Bound the edge-signal block: per-source caps, then a global cap.

    Sorts HIGH→MEDIUM→LOW, keeps at most MAX_FILING_SIGNALS filing-sourced and
    MAX_TRANSCRIPT_SIGNALS transcript-sourced signals, then truncates to
    max_total so neither module can flood the prompt.
    """
    order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
    signals = sorted(signals, key=lambda s: order.get(s.get("significance", "MEDIUM"), 2))
    capped: list[dict] = []
    filing_n = transcript_n = 0
    for s in signals:
        if s.get("source") == "transcript":
            if transcript_n >= MAX_TRANSCRIPT_SIGNALS:
                continue
            transcript_n += 1
        else:
            if filing_n >= MAX_FILING_SIGNALS:
                continue
            filing_n += 1
        capped.append(s)
    return capped[:max_total]


def signals_node(state: AgentState) -> dict:
    """Run deterministic analysis modules and inject signals into conversation."""
    ticker = state["ticker"]
    raw_signals = []
    try:
        from analysis.textdiff import compute as compute_text_deltas
        raw_signals.extend(compute_text_deltas(ticker))
    except Exception as exc:
        import sys
        print(f"[signals_node] textdiff error: {exc}", file=sys.stderr)
    try:
        from analysis.tone_drift import compute as compute_tone_deltas
        raw_signals.extend(compute_tone_deltas(ticker))
    except Exception as exc:
        import sys
        print(f"[signals_node] tone_drift error: {exc}", file=sys.stderr)
    signals = _cap_signals([s.model_dump() for s in raw_signals])

    if signals:
        msg = HumanMessage(content=_format_signals_message(signals))
        return {"edge_signals": signals, "messages": [msg]}
    return {"edge_signals": [], "messages": []}


def profile_evidence_node(state: AgentState) -> dict:
    """Inject deterministic profile evidence as parseable human messages."""
    from agent.company_profile import collect_profile_evidence

    payloads = collect_profile_evidence(state["ticker"], include_metrics=False)
    if not payloads:
        return {"profile_payloads": [], "messages": []}
    messages = [
        HumanMessage(
            content=(
                "== COMPANY PROFILE EVIDENCE "
                "(deterministic retrieval, evidence.v1 envelopes follow) =="
            )
        )
    ]
    messages.extend(HumanMessage(content=payload) for payload in payloads)
    return {"profile_payloads": payloads, "messages": messages}


def _pop_company_profile(data: dict) -> dict | None:
    """Remove the separately validated profile section from synthesis output."""
    return data.pop("company_profile", None)


def _finalize_synthesis_profile(
    state: AgentState,
    profile_section: dict,
    model: str | None,
) -> dict | None:
    """Finalize and persist a synthesized profile without risking the brief."""
    try:
        from agent.company_profile import finalize_profile_from_synthesis
        from storage import company_profiles

        all_payloads = list(state.get("messages") or []) + list(
            state.get("profile_payloads") or []
        )
        profile = finalize_profile_from_synthesis(
            state["ticker"], profile_section, all_payloads, model
        )
        company_profiles.save_profile(state["ticker"], profile)
        return profile
    except Exception as exc:
        import sys

        print(
            f"[synthesis] company profile finalization failed: {exc}",
            file=sys.stderr,
        )
        return None


def create_graph(config: Optional[RunConfig] = None):
    cfg = config or default_config()
    llm = make_chat_model(cfg)
    llm_with_tools = llm.bind_tools(TOOLS)

    def agent_node(state: AgentState) -> dict:
        system_block = build_system_message(cfg, SYSTEM_PROMPT)
        response = llm_with_tools.invoke([system_block] + state["messages"])
        return {"messages": [response]}

    def tool_node(state: AgentState) -> dict:
        last = state["messages"][-1]
        tool_calls = getattr(last, "tool_calls", None) or []
        if not tool_calls:
            return {"messages": [], "tool_round_count": state["tool_round_count"] + 1}
        with ThreadPoolExecutor(max_workers=min(8, len(tool_calls))) as ex:
            results = list(ex.map(_invoke_one, tool_calls))
        return {"messages": results, "tool_round_count": state["tool_round_count"] + 1}

    def partial_node(state: AgentState) -> dict:
        brief = _partial_brief(state, "Required evidence was unavailable or invalid.")
        return {
            "brief": brief,
            "brief_markdown": None,
            "synthesis_error": None,
            "coverage": brief["coverage"],
            "verification_report": brief["verification_report"],
        }

    def synthesis_node(state: AgentState) -> dict:
        try:
            llm_plain = make_chat_model(cfg, max_tokens=SYNTHESIS_MAX_TOKENS)
            # Main prompt — cached (ephemeral) on Anthropic. Keep this block stable
            # so the cache hit rate is preserved regardless of the chosen language.
            lang = state.get("language") or "English"
            extra_texts = [language_directive(lang)] if lang != "English" else None
            system_block = build_system_message(cfg, SYNTHESIS_STRUCTURED_PROMPT, extra_texts)
            synthesis_messages = [system_block] + state["messages"]
            # If cap was hit mid-round the last AIMessage may still carry tool_calls.
            # Anthropic rejects conversations where tool_use blocks have no matching
            # tool_result — insert stubs so the message history is valid.
            last_msg = synthesis_messages[-1]
            if getattr(last_msg, "tool_calls", None):
                stubs = [
                    ToolMessage(
                        tool_call_id=tc["id"],
                        name=tc["name"],
                        content=evidence_envelope(
                            tool=tc["name"], query=tc.get("args") or {}, status="ERROR",
                            message="Tool call interrupted at round cap.",
                            error_code="ROUND_CAP_INTERRUPTED",
                        ),
                    )
                    for tc in last_msg.tool_calls
                ]
                synthesis_messages = synthesis_messages + stubs
            if not isinstance(synthesis_messages[-1], HumanMessage):
                synthesis_messages = synthesis_messages + [
                    HumanMessage(content="Now produce the structured research brief as a JSON object.")
                ]
            chunks = []
            stop_reason = None
            for chunk in llm_plain.stream(synthesis_messages):
                metadata = getattr(chunk, "response_metadata", None) or {}
                stop_reason = (
                    metadata.get("stop_reason")
                    or metadata.get("finish_reason")
                    or stop_reason
                )
                text = chunk.content if isinstance(chunk.content, str) else ""
                if text:
                    chunks.append(text)
            if stop_reason in ("max_tokens", "length"):
                raise SynthesisTruncatedError(
                    "Synthesis output was truncated by the token limit "
                    f"(max_tokens={SYNTHESIS_MAX_TOKENS}); the brief JSON was incomplete."
                )
            raw = "".join(chunks)
            clean = _extract_json(raw)
            data = json.loads(clean)
            profile_section = _pop_company_profile(data)
            brief = BriefOutput.model_validate(data)
            brief_dict = apply_reliability(
                brief.model_dump(), evidence_payloads=state.get("messages", [])
            )
            brief_dict = attach_edge_signals(brief_dict, state.get("edge_signals"))
            brief_dict["language"] = lang
            coverage = _coverage_report(state.get("messages", []))
            evidence_coverage = brief_dict.get("evidence_coverage") or {}
            if evidence_coverage.get("verified", 0) < 1:
                note = (
                    "Synthesis contained no claim that passed deterministic "
                    "evidence verification."
                )
                brief_dict["evidence_notes"] = (
                    list(brief_dict.get("evidence_notes") or []) + [note]
                )[:6]

            generated_at = datetime.now(timezone.utc).isoformat()
            evidence_dates = []
            for message in state.get("messages", []):
                envelope = parse_evidence_envelope(message)
                if not envelope or envelope.get("status") != "OK":
                    continue
                evidence_dates.extend(
                    record.get("ref", {}).get("as_of")
                    for record in envelope.get("records") or []
                    if record.get("ref", {}).get("as_of")
                )
            brief_dict["schema_version"] = "brief.v2"
            brief_dict["generated_at"] = generated_at
            brief_dict["model"] = cfg.model
            brief_dict["data_as_of"] = max(evidence_dates) if evidence_dates else brief_dict.get("filing_date", "")
            brief_dict["coverage"] = coverage
            if coverage.get("status") != "COMPLETE" or evidence_coverage.get("status") != "VERIFIED":
                brief_dict["status"] = "PARTIAL"
            market = brief_dict.get("market_expectations") or {}
            brief_dict["display_policy"] = {
                "event_returns_aligned": (
                    market.get("event_aligned") is True
                    and market.get("event_comparison_allowed") is True
                ),
                "market_expectations_aligned": (
                    market.get("period_aligned") is True
                    and market.get("comparison_allowed") is True
                ),
                "aggregate_reliability_meaningful": False,
            }
            if isinstance(profile_section, dict) and profile_section:
                brief_dict["company_profile"] = _finalize_synthesis_profile(
                    state, profile_section, cfg.model
                )
            else:
                brief_dict["company_profile"] = None
            return {
                "brief": brief_dict,
                "brief_markdown": None,
                "synthesis_error": None,
                "coverage": coverage,
                "verification_report": brief_dict.get("verification_report"),
            }
        except Exception as exc:
            import sys
            print(f"[synthesis error] {exc}", file=sys.stderr)
            if isinstance(exc, SynthesisTruncatedError):
                reason = str(exc)
            else:
                reason = f"Synthesis failed validation: {exc}"
            partial = _partial_brief(state, reason)
            return {
                "brief": partial,
                "brief_markdown": None,
                "synthesis_error": str(exc),
                "coverage": partial["coverage"],
                "verification_report": partial["verification_report"],
            }

    builder = StateGraph(AgentState)
    builder.add_node("signals", signals_node)
    builder.add_node("profile_evidence", profile_evidence_node)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", tool_node)
    builder.add_node("nudge", nudge_node)
    builder.add_node("synthesis", synthesis_node)
    builder.add_node("partial", partial_node)
    builder.set_entry_point("signals")
    builder.add_edge("signals", "profile_evidence")
    builder.add_edge("profile_evidence", "agent")
    builder.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", "nudge": "nudge", "synthesis": "synthesis", "partial": "partial"},
    )
    builder.add_edge("tools", "agent")
    builder.add_edge("nudge", "agent")
    builder.add_edge("synthesis", END)
    builder.add_edge("partial", END)
    return builder.compile()


def run_brief(ticker: str, language: str = "English", config: Optional[RunConfig] = None) -> Optional[dict]:
    graph = create_graph(config)
    final = graph.invoke({
        "ticker": ticker.upper(),
        "messages": [HumanMessage(content=f"Generate a research brief for {ticker.upper()}.")],
        "tool_round_count": 0,
        "nudge_fired": False,
        "edge_signals": None,
        "profile_payloads": None,
        "language": language,
        "brief": None,
        "brief_markdown": None,
        "synthesis_error": None,
        "coverage": None,
        "verification_report": None,
    })
    return final.get("brief")