Step 2. Define the Agents SDK-backed analyst

The example agent performs acquisition diligence on a fictional SaaS company being reviewed as a possible acquisition target. The case materials contain both structured exports and management narratives. Some sources agree, some conflict, and some important claims are only partially supported. That gives us a realistic reason to improve the harness over time.

The agent answers questions for an investment team using only the supplied company data. It should prefer structured financial evidence over narrative summaries when they disagree, preserve uncertainty when evidence is missing, and leave behind artifacts that another reviewer can inspect.

The OpenAI Agents SDK provides the managed runner, sandbox execution, model settings, and tracing hooks this workflow needs. Together, the prompt, tools, routing rules, output requirements, and validation checks form the current agent harness.

Artifacts generated by the agent

Artifact	Why the agent writes it
summary_answer.md	The concise answer returned to the user
investment_memo.md	A fuller review artifact for diligence readers
risk_register.json	Structured risks with evidence that downstream systems can inspect
open_questions.md	Missing evidence or unresolved questions that should stay visible
citations.json	A machine-readable link from claims to source files
evidence_table.csv	A tabular audit trail of claims and supporting sources

These artifacts keep the work reviewable by preserving supporting evidence, unresolved questions, and required files alongside the final answer.

Failure modes to watch for

This notebook is designed to surface failures such as:

• Treating management narrative as an official metric when the structured exports disagree
• Reporting an unsupported NRR estimate as if finance had validated it
• Collapsing parent-account concentration into a weaker legal-entity view
• Saying “SOC 2 complete” when the evidence only supports Type I
• Producing a polished answer while leaving citations, risk files, or evidence artifacts incomplete

Define the harness schema

Start with small data structures for the model settings and promoted agent configuration. These make the harness explicit so later optimization can target more than prompt wording.

@dataclass(frozen=True)
class ModelSettings:
    agent_model: str
    reasoning_effort: str

@dataclass(frozen=True)
class AgentConfig:
    version: str
    system_prompt: str
    model_settings: ModelSettings
    tool_policy: dict[str, Any]
    eval_metadata: dict[str, Any]
    path: Path = field(default_factory=lambda: Path("notebook_defined_agent_config"))

    @property
    def required_artifacts(self) -> list[str]:
        return self.tool_policy["required_artifacts"]

    def build_instructions(self) -> str:
        return "\n\n".join([
            self.system_prompt,
            format_policy_section("Tool policy", self.tool_policy),
            f"Runtime config:\n- Config version: `{self.version}`.\n- Treat this config as the promoted runtime contract.\n- Do not modify the runtime config during the run.",
        ]) + "\n"

def format_policy_section(title: str, policy: dict[str, Any]) -> str:
    lines = [f"{title}:"]
    for key, value in policy.items():
        lines.extend(format_policy_value(key, value))
    return "\n".join(lines)

def format_policy_value(key: str, value: Any, indent: int = 0) -> list[str]:
    prefix = "  " * indent
    if isinstance(value, dict):
        lines = [f"{prefix}- {key}:"]
        for child_key, child_value in value.items():
            lines.extend(format_policy_value(child_key, child_value, indent + 1))
        return lines
    if isinstance(value, list):
        lines = [f"{prefix}- {key}:"]
        for item in value:
            if isinstance(item, dict):
                lines.append(f"{prefix}  -")
                for child_key, child_value in item.items():
                    lines.extend(format_policy_value(child_key, child_value, indent + 2))
            else:
                lines.append(f"{prefix}  - {item}")
        return lines
    return [f"{prefix}- {key}: {value}"]

Configure instructions and policies

The system prompt states the evidence rules, the tool policy defines what the agent may read and write, and the eval metadata records which version of the harness is currently promoted.

SYSTEM_PROMPT = """
You are a diligence analyst reviewing a synthetic company dataroom.

Evidence scope:
- Use only files under `data/`.
- Do not use outside knowledge or assumptions.
- Prefer structured CSV/JSON exports over narrative files when they conflict.

Runtime tools:
- The sandbox starts in the mounted workspace root. Use workspace-relative paths such as `data/...` and `outputs/...`; when running shell commands, omit `workdir` or use a relative path only. Never pass absolute temporary paths.
- `data/tools/check_evidence_coverage.py`: use this before finalizing answers with material claims. Create a JSON list of claims with `claim`, `claim_type`, and `citations`, then run `python data/tools/check_evidence_coverage.py --claims-json <path> --dataset-root data --output outputs/evidence_coverage.json`.
- `data/tools/validate_output_contract.py`: run this after writing the required artifacts and before final response with `python data/tools/validate_output_contract.py --outputs outputs --dataset-root data --output outputs/output_contract_validation.json`.
- If either tool reports unsupported claims, missing citations, missing files, malformed JSON, or empty artifacts, revise the answer/artifacts before finalizing. If the evidence is unavailable, say the claim is unknown or unsupported.

Citation rules:
- Every material claim must cite one or more source filenames.
- Cite filenames exactly as workspace-relative paths, for example `financials/arr_bridge.csv`.
- Do not cite files that do not support the claim.

Unknown-handling rules:
- If evidence is missing, state that the answer is unknown or unsupported.
- Never fabricate missing numbers.
- If evidence conflicts, state the conflict explicitly instead of reconciling silently.

Output rules:
- Write `outputs/summary_answer.md`.
- Write `outputs/investment_memo.md`.
- Write `outputs/risk_register.json`.
- Write `outputs/open_questions.md`.
- Write `outputs/citations.json`.
- Write `outputs/evidence_table.csv`.
""".strip()

MODEL_SETTINGS = {
    "agent_model": AGENT_MODEL,
    "reasoning_effort": "medium",
}

TOOL_POLICY = {
    "allowed_data_root": "data",
    "writable_output_root": "outputs",
    "required_artifacts": [
        "summary_answer.md",
        "investment_memo.md",
        "risk_register.json",
        "open_questions.md",
        "citations.json",
        "evidence_table.csv",
    ],
    "evidence_preference": [
        "Prefer structured CSV or JSON exports over narrative summaries when sources conflict.",
        "Treat board materials as useful narrative evidence, not the final system of record for metrics.",
        "Surface unresolved conflicts instead of silently reconciling them.",
    ],
    "runtime_tools": [
        {
            "path": "data/tools/check_evidence_coverage.py",
            "purpose": "Audit drafted material claims against cited dataroom files before final answer.",
            "recommended_command": "python data/tools/check_evidence_coverage.py --claims-json outputs/claim_audit_input.json --dataset-root data --output outputs/evidence_coverage.json",
        },
        {
            "path": "data/tools/validate_output_contract.py",
            "purpose": "Validate required output artifacts, JSON shape, and citation/source file references.",
            "recommended_command": "python data/tools/validate_output_contract.py --outputs outputs --dataset-root data --output outputs/output_contract_validation.json",
        },
    ],
    "unknown_handling": [
        "Say unknown or unsupported when a metric is absent.",
        "Do not infer missing values from adjacent metrics.",
        "Keep facts, inferences, and open questions separate.",
    ],
    "mutation_policy": [
        "Write only to the configured outputs directory.",
        "Do not modify dataroom inputs.",
        "Do not modify runtime agent configuration during a run.",
    ],
}

EVAL_METADATA = {
    "version": "v001",
    "status": "promoted",
    "created_by": "manual_baseline",
    "promotion_gate": "manual_review",
    "description": "Baseline diligence analyst config with strict dataroom grounding, citation, unknown-handling, and artifact rules.",
}

agent_config = AgentConfig(
    version=EVAL_METADATA["version"],
    system_prompt=SYSTEM_PROMPT,
    model_settings=ModelSettings(**MODEL_SETTINGS),
    tool_policy=TOOL_POLICY,
    eval_metadata=EVAL_METADATA,
)

Inspect the agent config

This compact view shows the promoted config version, the selected models, the required artifacts, and the runtime tools the agent can use.

required_artifacts_md = "\n".join(
    f"- `{artifact}`" for artifact in agent_config.required_artifacts
)
runtime_tools_md = "\n".join(
    f"- `{tool['path']}` — {tool['purpose']}"
    for tool in agent_config.tool_policy["runtime_tools"]
)

display(Markdown(f"""
### Agent config summary

- **Version:** `{agent_config.version}`
- **Agent model:** `{agent_config.model_settings.agent_model}`
- **Reasoning effort:** `{agent_config.model_settings.reasoning_effort}`

**Required artifacts**
{required_artifacts_md}

**Runtime tools**
{runtime_tools_md}
"""))

Agent config summary

• Version: v001
• Agent model: gpt-5.5
• Reasoning effort: medium

Required artifacts

• summary_answer.md
• investment_memo.md
• risk_register.json
• open_questions.md
• citations.json
• evidence_table.csv

Runtime tools

• data/tools/check_evidence_coverage.py — Audit drafted material claims against cited dataroom files before final answer.
• data/tools/validate_output_contract.py — Validate required output artifacts, JSON shape, and citation/source file references.

Add validation tools

The next helpers create two local tools inside the workspace: one checks whether drafted claims cite real dataroom files, and the other verifies that the required output artifacts exist and have the expected shape. The code is hidden by default to save space, but you can expand it if you want to inspect the implementation.

CHECK_EVIDENCE_COVERAGE = r'''#!/usr/bin/env python3

import argparse
import json
from pathlib import Path

def main() -> None:
    parser = argparse.ArgumentParser(description="Audit whether drafted claims cite existing dataroom files.")
    parser.add_argument("--claims-json", type=Path, required=True)
    parser.add_argument("--dataset-root", type=Path, default=Path("data"))
    parser.add_argument("--output", type=Path, default=Path("outputs/evidence_coverage.json"))
    args = parser.parse_args()

    claims = json.loads(args.claims_json.read_text(encoding="utf-8"))
    if not isinstance(claims, list):
        raise ValueError("--claims-json must contain a JSON list of claim objects")

    result = check_evidence_coverage(claims, args.dataset_root)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(result, indent=2))

def check_evidence_coverage(claims: list[dict], dataset_root: Path) -> dict:
    supported = []
    unsupported = []
    missing_citations = []

    for raw in claims:
        claim = str(raw.get("claim") or "").strip()
        claim_type = str(raw.get("claim_type") or "claim")
        citations = [str(item).strip().removeprefix("data/") for item in raw.get("citations") or [] if str(item).strip()]
        row = {"claim": claim, "claim_type": claim_type, "citations": citations}
        if not citations:
            missing_citations.append({**row, "issue": "No citation provided."})
            continue
        missing = [citation for citation in citations if not (dataset_root / citation).exists()]
        if missing:
            unsupported.append({**row, "issue": f"Missing cited file(s): {', '.join(missing)}"})
        else:
            supported.append(row)

    return {
        "supported_claims": supported,
        "unsupported_claims": unsupported,
        "missing_citations": missing_citations,
        "recommended_caveats": [
            "Add valid source filenames or mark unsupported claims as unknown before final answer."
        ],
        "passed": not unsupported and not missing_citations,
    }

if __name__ == "__main__":
    main()
'''

VALIDATE_OUTPUT_CONTRACT = r'''#!/usr/bin/env python3

import argparse
import csv
import json
from pathlib import Path

REQUIRED_FILES = [
    "summary_answer.md",
    "investment_memo.md",
    "risk_register.json",
    "open_questions.md",
    "citations.json",
    "evidence_table.csv",
]

def main() -> None:
    parser = argparse.ArgumentParser(description="Validate diligence output artifacts before final answer.")
    parser.add_argument("--outputs", type=Path, default=Path("outputs"))
    parser.add_argument("--dataset-root", type=Path, default=Path("data"))
    parser.add_argument("--output", type=Path, default=Path("outputs/output_contract_validation.json"))
    args = parser.parse_args()

    result = validate_output_contract(args.outputs, args.dataset_root)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(result, indent=2))

def validate_output_contract(outputs: Path, dataset_root: Path) -> dict:
    issues = []
    for filename in REQUIRED_FILES:
        path = outputs / filename
        if not path.exists():
            issues.append({"file": filename, "issue": "missing required artifact"})
        elif path.stat().st_size == 0:
            issues.append({"file": filename, "issue": "empty required artifact"})

    risks = _read_json(outputs / "risk_register.json", default=[])
    citations = _read_json(outputs / "citations.json", default=[])
    if not isinstance(risks, list):
        issues.append({"file": "risk_register.json", "issue": "must be a JSON list"})
        risks = []
    if not isinstance(citations, list):
        issues.append({"file": "citations.json", "issue": "must be a JSON list"})
        citations = []

    for index, risk in enumerate(risks):
        evidence = risk.get("evidence") if isinstance(risk, dict) else None
        if not evidence:
            issues.append({"file": "risk_register.json", "risk_index": index, "issue": "risk lacks evidence"})
            continue
        missing = [str(item).removeprefix("data/") for item in evidence if not (dataset_root / str(item).removeprefix("data/")).exists()]
        if missing:
            issues.append({"file": "risk_register.json", "risk_index": index, "issue": f"missing evidence file(s): {', '.join(missing)}"})

    for index, citation in enumerate(citations):
        sources = citation.get("sources") if isinstance(citation, dict) else None
        if not sources:
            issues.append({"file": "citations.json", "citation_index": index, "issue": "citation lacks sources"})
            continue
        missing = [str(item).removeprefix("data/") for item in sources if not (dataset_root / str(item).removeprefix("data/")).exists()]
        if missing:
            issues.append({"file": "citations.json", "citation_index": index, "issue": f"missing source file(s): {', '.join(missing)}"})

    try:
        with (outputs / "evidence_table.csv").open(newline="", encoding="utf-8") as handle:
            rows = list(csv.DictReader(handle))
        if rows and not {"claim_id", "claim", "sources"}.issubset(rows[0].keys()):
            issues.append({"file": "evidence_table.csv", "issue": "must include claim_id, claim, and sources columns"})
    except FileNotFoundError:
        pass

    return {"passed": not issues, "issues": issues, "required_files": REQUIRED_FILES}

def _read_json(path: Path, default):
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        return {"error": str(exc)}

if __name__ == "__main__":
    main()
'''

def write_runtime_tools(dataset_dir: Path) -> list[str]:
    tools_dir = dataset_dir / "tools"
    tools_dir.mkdir(parents=True, exist_ok=True)
    files = {
        "check_evidence_coverage.py": CHECK_EVIDENCE_COVERAGE,
        "validate_output_contract.py": VALIDATE_OUTPUT_CONTRACT,
    }
    written: list[str] = []
    for filename, content in files.items():
        path = tools_dir / filename
        path.write_text(content, encoding="utf-8")
        path.chmod(0o755)
        written.append(str(path.relative_to(dataset_dir)))
    return written

Build each user turn

The prompt builder adds task-specific guidance only when it is needed, such as memo formatting, separate risk categories, or strict handling for unsupported NRR claims.

def build_user_prompt(question: str, agent_config: Any | None = None) -> str:
    config_line = ""
    if agent_config is not None:
        config_line = f"\nActive agent config: `{agent_config.version}` from `{agent_config.path}`.\n"
    memo_instruction = ""
    if _asks_for_memo(question):
        memo_instruction = (
            "\nThe user asked for a memo-style deliverable. Return the memo content inline in "
            "your final answer and also write the required output artifacts. Do not answer only "
            "with a status update or artifact path list.\n"
        )
    risk_category_instruction = ""
    if _asks_for_top_risk_categories(question):
        risk_category_instruction = (
            "\nStructure the final answer with separate sections for Financial, Legal, and "
            "Customer concentration risks. Do not collapse customer concentration into the "
            "financial category.\n"
        )
    unsupported_metric_instruction = ""
    if _asks_for_net_revenue_retention(question):
        unsupported_metric_instruction = (
            "\nFor net revenue retention, report the metric only if the dataroom directly "
            "provides NRR/net revenue retention. Do not derive or estimate an NRR percentage "
            "from ARR bridge components unless the user explicitly asks for an estimate. If "
            "the metric is absent, say it is unknown or unsupported, cite the searched "
            "source files, and separate missing evidence from any directional inference.\n"
        )
    return f"""
Answer this diligence question using only the mounted dataroom:

{question}
{config_line}
{memo_instruction}
{risk_category_instruction}
{unsupported_metric_instruction}
Also write the required output artifacts. Keep the answer concise, grounded, and citation-heavy.
Use workspace-relative paths for shell commands and omit `workdir`; do not pass absolute temporary paths.
"""

def _asks_for_memo(question: str) -> bool:
    lower = question.lower()
    return "memo" in lower or "ic-style" in lower or "investment committee" in lower

def _asks_for_top_risk_categories(question: str) -> bool:
    lower = question.lower()
    return all(term in lower for term in ("financial", "legal", "customer")) and "risk" in lower

def _asks_for_net_revenue_retention(question: str) -> bool:
    lower = question.lower()
    return "net revenue retention" in lower or "nrr" in lower

Export traces for later optimization

The local exporter converts Agents SDK events into the OpenTelemetry-style JSONL that HALO can read later. It is implementation-heavy, so the code stays collapsed by default.

Configure the trace exporter

Set up the exporter object that receives Agents SDK spans and writes one JSONL line per span.

EXPORT_SCHEMA_VERSION = 1

OBSERVATION_KIND_BY_TYPE = {
    "agent": "AGENT",
    "generation": "LLM",
    "response": "LLM",
    "function": "TOOL",
    "mcp_tools": "TOOL",
    "handoff": "CHAIN",
    "guardrail": "GUARDRAIL",
    "custom": "SPAN",
    "task": "SPAN",
    "turn": "SPAN",
    "transcription": "SPAN",
    "speech": "SPAN",
    "speech_group": "SPAN",
}

@dataclass(frozen=True)
class HaloExportContext:
    project_id: str
    service_name: str
    service_version: str | None = None
    deployment_environment: str | None = None
    extra_resource_attributes: Mapping[str, Any] | None = None

def setup_halo_tracing(
    path: str | Path,
    *,
    project_id: str = "synthetic-dataroom-agent",
    service_name: str = "financial-diligence-analyst",
    service_version: str | None = None,
    deployment_environment: str | None = None,
    extra_resource_attributes: Mapping[str, Any] | None = None,
):
    from agents import set_trace_processors

    trace_path = Path(path)
    trace_path.parent.mkdir(parents=True, exist_ok=True)
    processor = HaloJsonlTraceProcessor(
        trace_path,
        ctx=HaloExportContext(
            project_id=project_id,
            service_name=service_name,
            service_version=service_version,
            deployment_environment=deployment_environment,
            extra_resource_attributes=extra_resource_attributes,
        ),
    )
    # Use only the local exporter for this cookbook workflow.
    # Hosted trace ingestion may be unavailable in some environments (for example ZDR orgs).
    set_trace_processors([processor])
    return processor

class HaloJsonlTraceProcessor:
    def __init__(self, path: Path, *, ctx: HaloExportContext):
        self._path = path
        self._ctx = ctx
        self._lock = threading.Lock()
        self._handle = path.open("a", encoding="utf-8")
        self._trace_meta: dict[str, tuple[str | None, str | None, dict[str, Any]]] = {}

    def on_trace_start(self, trace) -> None:  # noqa: ANN001
        data = trace.export() or {}
        trace_id = _strip_prefix(data.get("id"), "trace_") or ""
        metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
        self._trace_meta[trace_id] = (
            data.get("workflow_name"),
            data.get("group_id"),
            metadata,
        )

    def on_trace_end(self, trace) -> None:  # noqa: ANN001
        data = trace.export() or {}
        trace_id = _strip_prefix(data.get("id"), "trace_") or ""
        self._trace_meta.pop(trace_id, None)

    def on_span_start(self, span) -> None:  # noqa: ANN001
        return None

    def on_span_end(self, span) -> None:  # noqa: ANN001
        exported = span.export() or {}
        trace_id = _strip_prefix(exported.get("trace_id"), "trace_") or ""
        workflow_name, group_id, trace_metadata = self._trace_meta.get(trace_id, (None, None, {}))
        line = span_to_halo_jsonl_line(
            span,
            ctx=self._ctx,
            workflow_name=workflow_name,
            group_id=group_id,
            trace_metadata=trace_metadata,
        )
        encoded = json.dumps(line, separators=(",", ":"), ensure_ascii=False, default=str)
        with self._lock:
            self._handle.write(encoded)
            self._handle.write("\n")

    def shutdown(self) -> None:
        with self._lock:
            try:
                self._handle.flush()
                self._handle.close()
            except Exception:
                pass

    def force_flush(self) -> None:
        with self._lock:
            self._handle.flush()

Map SDK spans into HALO-readable fields

These helpers translate each SDK span type into the attributes HALO will inspect later.

def span_to_halo_jsonl_line(
    span,
    *,
    ctx: HaloExportContext,
    workflow_name: str | None = None,
    group_id: str | None = None,
    trace_metadata: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    raw = span.export() or {}
    span_data = raw.get("span_data") or {}
    span_type = str(span_data.get("type") or "custom")
    error = raw.get("error")
    resource_attributes: dict[str, Any] = {"service.name": ctx.service_name}
    if ctx.service_version:
        resource_attributes["service.version"] = ctx.service_version
    if ctx.deployment_environment:
        resource_attributes["deployment.environment"] = ctx.deployment_environment
    if ctx.extra_resource_attributes:
        resource_attributes.update(ctx.extra_resource_attributes)

    attributes, projection = _attributes_for_span_type(span_type, span_data)
    if workflow_name:
        attributes["agent.workflow.name"] = workflow_name
    if group_id:
        attributes["agent.workflow.group_id"] = group_id
    for key, value in (trace_metadata or {}).items():
        if _json_safe(value):
            attributes[f"agent.trace_metadata.{key}"] = value
        else:
            attributes[f"agent.trace_metadata.{key}"] = _json(value)

    attributes.update(
        {
            "inference.export.schema_version": EXPORT_SCHEMA_VERSION,
            "inference.project_id": ctx.project_id,
            "inference.observation_kind": OBSERVATION_KIND_BY_TYPE.get(span_type, "SPAN"),
            "inference.llm.provider": projection.get("llm_provider"),
            "inference.llm.model_name": projection.get("llm_model_name"),
            "inference.llm.input_tokens": projection.get("input_tokens"),
            "inference.llm.output_tokens": projection.get("output_tokens"),
            "inference.llm.cost.total": projection.get("cost_total"),
            "inference.user_id": projection.get("user_id"),
            "inference.session_id": group_id,
            "inference.agent_name": projection.get("agent_name") or "",
        }
    )

    return {
        "trace_id": _strip_prefix(raw.get("trace_id"), "trace_") or "",
        "span_id": _strip_prefix(raw.get("id"), "span_") or "",
        "parent_span_id": _strip_prefix(raw.get("parent_id"), "span_") or "",
        "trace_state": "",
        "name": _span_name(span_type, span_data),
        "kind": _span_kind(span_type),
        "start_time": _to_otlp_timestamp(raw.get("started_at")),
        "end_time": _to_otlp_timestamp(raw.get("ended_at")),
        "status": {
            "code": "STATUS_CODE_ERROR" if error else "STATUS_CODE_OK",
            "message": str((error or {}).get("message") or ""),
        },
        "resource": {"attributes": resource_attributes},
        "scope": {"name": "openai-agents-sdk", "version": _sdk_version()},
        "attributes": {key: value for key, value in attributes.items() if value is not None},
    }

def _attributes_for_span_type(
    span_type: str,
    data: Mapping[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
    if span_type == "agent":
        return _agent_attrs(data)
    if span_type == "generation":
        return _generation_attrs(data)
    if span_type == "response":
        return _response_attrs(data)
    if span_type == "function":
        return _function_attrs(data)
    if span_type == "mcp_tools":
        return _mcp_tools_attrs(data)
    if span_type == "handoff":
        return _handoff_attrs(data)
    if span_type == "guardrail":
        return _guardrail_attrs(data)
    return _custom_attrs(span_type, data)

def _agent_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    name = data.get("name") or ""
    return _drop_none(
        {
            "openinference.span.kind": "AGENT",
            "agent.name": name,
            "agent.handoffs": _json(data.get("handoffs")),
            "agent.tools": _json(data.get("tools")),
            "agent.output_type": data.get("output_type"),
        }
    ), {"agent_name": name}

def _generation_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    usage = data.get("usage") or {}
    input_messages = data.get("input") or []
    output_messages = data.get("output") or []
    attrs: dict[str, Any] = {
        "openinference.span.kind": "LLM",
        "llm.provider": "openai",
        "llm.model_name": data.get("model"),
        "llm.invocation_parameters": _json(data.get("model_config")),
        "llm.input_messages": _json(list(input_messages)),
        "llm.output_messages": _json(list(output_messages)),
        "llm.token_count.prompt": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
        "llm.token_count.completion": _int(
            usage.get("output_tokens") or usage.get("completion_tokens")
        ),
        "llm.token_count.total": _int(usage.get("total_tokens")),
    }
    attrs.update(_expand_messages("llm.input_messages", input_messages))
    attrs.update(_expand_messages("llm.output_messages", output_messages))
    return _drop_none(attrs), {
        "llm_provider": "openai",
        "llm_model_name": data.get("model"),
        "input_tokens": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
        "output_tokens": _int(usage.get("output_tokens") or usage.get("completion_tokens")),
    }

def _response_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    usage = data.get("usage") or {}
    return _drop_none(
        {
            "openinference.span.kind": "LLM",
            "llm.provider": "openai",
            "llm.response.id": data.get("response_id"),
            "llm.token_count.prompt": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
            "llm.token_count.completion": _int(
                usage.get("output_tokens") or usage.get("completion_tokens")
            ),
            "llm.token_count.total": _int(usage.get("total_tokens")),
        }
    ), {
        "llm_provider": "openai",
        "input_tokens": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
        "output_tokens": _int(usage.get("output_tokens") or usage.get("completion_tokens")),
    }

def _function_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    return _drop_none(
        {
            "openinference.span.kind": "TOOL",
            "tool.name": data.get("name"),
            "input.value": data.get("input"),
            "output.value": data.get("output"),
            "mcp.data": _json(data.get("mcp_data")),
        }
    ), {}

def _mcp_tools_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    return _drop_none(
        {
            "openinference.span.kind": "TOOL",
            "mcp.server": data.get("server"),
            "mcp.tools.listed": _json(data.get("result")),
        }
    ), {}

def _handoff_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    return _drop_none(
        {
            "openinference.span.kind": "CHAIN",
            "agent.handoff.from": data.get("from_agent"),
            "agent.handoff.to": data.get("to_agent"),
        }
    ), {"agent_name": data.get("to_agent")}

def _guardrail_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    return _drop_none(
        {
            "openinference.span.kind": "GUARDRAIL",
            "guardrail.name": data.get("name"),
            "guardrail.triggered": bool(data.get("triggered")),
        }
    ), {}

def _custom_attrs(span_type: str, data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    attrs: dict[str, Any] = {
        "openinference.span.kind": "CHAIN",
        "sdk.span.type": span_type,
    }
    if data.get("name"):
        attrs["sdk.span.name"] = data.get("name")
    payload = data.get("data") or {}
    if isinstance(payload, Mapping):
        for key, value in payload.items():
            attrs[f"sdk.data.{key}"] = value if _json_safe(value) else _json(value)
    if "usage" in data:
        attrs["llm.token_count.total"] = _int((data.get("usage") or {}).get("total_tokens"))
    return _drop_none(attrs), {}

Normalize helper values

The final helpers keep IDs, timestamps, and serialized values consistent across exported spans.

def _strip_prefix(value: Any, prefix: str) -> str | None:
    if not value:
        return None
    text = str(value)
    return text[len(prefix) :] if text.startswith(prefix) else text

def _to_otlp_timestamp(value: str | None) -> str:
    if not value:
        return ""
    parsed = datetime.fromisoformat(value)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    parsed = parsed.astimezone(timezone.utc)
    return parsed.strftime("%Y-%m-%dT%H:%M:%S.") + f"{parsed.microsecond:06d}000Z"

def _span_kind(span_type: str) -> str:
    return "SPAN_KIND_CLIENT" if span_type in {"generation", "response"} else "SPAN_KIND_INTERNAL"

def _span_name(span_type: str, data: Mapping[str, Any]) -> str:
    if data.get("name"):
        return f"{span_type}.{data['name']}"
    if data.get("model"):
        return f"{span_type}.{data['model']}"
    return span_type

def _expand_messages(prefix: str, messages: Iterable[Mapping[str, Any]]) -> dict[str, Any]:
    attrs: dict[str, Any] = {}
    for index, message in enumerate(messages or []):
        if not isinstance(message, Mapping):
            continue
        role = message.get("role")
        content = message.get("content")
        if role is not None:
            attrs[f"{prefix}.{index}.message.role"] = role
        if isinstance(content, str):
            attrs[f"{prefix}.{index}.message.content"] = content
        elif content is not None:
            attrs[f"{prefix}.{index}.message.content"] = _json(content)
        for tool_index, tool_call in enumerate(message.get("tool_calls") or []):
            function = (tool_call or {}).get("function") or {}
            attrs[f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.id"] = (
                tool_call or {}
            ).get("id")
            attrs[
                f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.function.name"
            ] = function.get("name")
            attrs[
                f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.function.arguments"
            ] = function.get("arguments")
        if message.get("tool_call_id"):
            attrs[f"{prefix}.{index}.message.tool_call_id"] = message["tool_call_id"]
        if message.get("name"):
            attrs[f"{prefix}.{index}.message.name"] = message["name"]
    return {key: value for key, value in attrs.items() if value is not None}

def _json(value: Any) -> str | None:
    if value is None:
        return None
    return json.dumps(value, default=str, separators=(",", ":"))

def _json_safe(value: Any) -> bool:
    return isinstance(value, (str, int, float, bool)) or value is None

def _int(value: Any) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None

def _drop_none(values: Mapping[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in values.items() if value is not None}

def _sdk_version() -> str:
    try:
        return version("openai-agents")
    except Exception:
        return "unknown"

Run the SDK agent

run_sdk_agent() calls the Agents SDK runner directly while handling the repeated setup around each traced run: mounting the data, attaching tracing, executing the agent, and collecting the output artifacts.

async def run_sdk_agent(
    dataset_dir: Path,
    output_dir: Path,
    question: str,
    model: str,
    agent_config: AgentConfig,
    trace_id: str | None = None,
    trace_metadata: dict[str, Any] | None = None,
    halo_trace_path: str | Path | None = None,
    halo_project_id: str = "financial_diligence_analyst_optimization_context",
) -> str:
    from agents import ModelSettings as SDKModelSettings
    from agents import Runner, custom_span, flush_traces, trace
    from agents.run import RunConfig
    from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
    from agents.sandbox.entries import Dir, LocalDir
    from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient
    from openai.types.shared import Reasoning

    output_dir.mkdir(parents=True, exist_ok=True)
    with staged_dataset_mount(dataset_dir) as staged_dataset_dir:
        write_runtime_manifest(staged_dataset_dir)
        reasoning = Reasoning(effort=agent_config.model_settings.reasoning_effort)
        agent = SandboxAgent(
            name="Synthetic dataroom diligence analyst",
            model=model,
            model_settings=SDKModelSettings(reasoning=reasoning),
            instructions=agent_config.build_instructions(),
            default_manifest=Manifest(
                entries={
                    "data": LocalDir(src=staged_dataset_dir),
                    "outputs": Dir(),
                }
            ),
        )
        client = UnixLocalSandboxClient()
        session = None
        halo_processor = None
        if halo_trace_path is not None:
            halo_processor = setup_halo_tracing(
                halo_trace_path,
                project_id=halo_project_id,
                service_version=agent_config.version,
                deployment_environment="notebook" if trace_metadata else None,
                extra_resource_attributes={
                    "agent.config.version": agent_config.version,
                    "agent.config.path": str(agent_config.path),
                },
            )
        trace_context = (
            trace(
                workflow_name="Synthetic dataroom diligence",
                trace_id=trace_id,
                metadata=trace_metadata,
            )
            if trace_id
            else None
        )
        if trace_context is not None:
            trace_context.__enter__()
        try:
            with custom_span(
                "sandbox_workspace",
                {
                    "tool.name": "sandbox_workspace",
                    "tool.input": {
                        "mounted": "data",
                        "writable": "outputs",
                        "dataset_dir": str(dataset_dir),
                        "staged_dataset_dir": str(staged_dataset_dir),
                        "agent_config": str(agent_config.path),
                        "agent_config_version": agent_config.version,
                    },
                },
                disabled=trace_context is None,
            ):
                with custom_span(
                    "agent_config",
                    {
                        "tool.name": "agent_config",
                        "tool.input": {
                            "version": agent_config.version,
                            "required_artifacts": agent_config.required_artifacts,
                        },
                    },
                    disabled=trace_context is None,
                ):
                    pass
                session = await client.create(manifest=agent.default_manifest)
            async with session:
                result = await Runner.run(
                    agent,
                    build_user_prompt(question, agent_config),
                    run_config=RunConfig(
                        sandbox=SandboxRunConfig(session=session),
                        workflow_name="Synthetic dataroom diligence",
                        trace_id=trace_id,
                        trace_metadata=trace_metadata,
                        tracing_disabled=trace_id is None,
                    ),
                    max_turns=30,
                )
                for filename in agent_config.required_artifacts:
                    try:
                        with custom_span(
                            "artifact_write",
                            {
                                "tool.name": "artifact_write",
                                "tool.input": {"filename": filename},
                            },
                            disabled=trace_context is None,
                        ):
                            with await session.read(Path("outputs") / filename) as handle:
                                (output_dir / filename).write_bytes(handle.read())
                    except Exception:
                        continue
                return str(result.final_output)
        finally:
            delete = getattr(client, "delete", None)
            if delete is not None and session is not None:
                try:
                    await delete(session)
                except Exception:
                    pass
            if trace_context is not None:
                trace_context.__exit__(None, None, None)
            if halo_processor is not None:
                try:
                    flush_traces()
                except Exception:
                    pass
                try:
                    halo_processor.shutdown()
                except Exception:
                    pass

@contextmanager
def staged_dataset_mount(dataset_dir: Path) -> Iterator[Path]:
    """Prepare a writable SDK mount copy without mutating the source dataroom."""
    with tempfile.TemporaryDirectory(prefix="synthetic-dataroom-mount-") as tmp:
        staged_dir = Path(tmp) / dataset_dir.name
        shutil.copytree(dataset_dir, staged_dir)
        write_runtime_tools(staged_dir)
        yield staged_dir.resolve()

def write_runtime_manifest(dataset_dir: Path) -> None:
    manifest = {
        "runtime_scope": "sdk_agent_visible_dataroom",
        "files": sorted(
            str(path.relative_to(dataset_dir))
            for path in dataset_dir.rglob("*")
            if path.is_file() and path.name != "manifest.json"
        ),
    }
    (dataset_dir / "manifest.json").write_text(
        json.dumps(manifest, indent=2) + "\n",
        encoding="utf-8",
    )