Skip to content

Python API Reference

Top-level exports

import notarize

notarize

notarize - Canonical trace format and verifier for agent execution attestation

ClosedLoopError

Bases: ValueError

Raised when the gate refuses empty or unusable traces.

GateOutcome(ok, verdict, reason, exit_code, verification=None, trace_id=None, failed_step_indices=(), degraded_step_indices=(), silent_success=False) dataclass

Result of a closed-loop read of a notarize trace.

Attributes:

Name Type Description
ok bool

True only when verification would let a pipeline continue.
verdict str

PASS, FAIL, or FAIL_LOUD.
reason str

Human-readable explanation (always non-empty).
exit_code int

0 for PASS, 1 for FAIL (tamper/invalid/degraded), 2 for FAIL_LOUD.
verification VerificationResult | None

Underlying :class:VerificationResult when scoring ran.
trace_id str | None

Trace identifier when available.
failed_step_indices tuple[int, ...]

Steps with hard-failure results.
degraded_step_indices tuple[int, ...]

Steps with degraded/partial results.
silent_success bool

True when claimed success conflicts with step outcomes.

to_dict()

Serialise for JSON reports (eagle-eyes dogfood, CI artifacts).

Source code in src/notarize/closed_loop.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def to_dict(self) -> dict[str, Any]:
    """Serialise for JSON reports (eagle-eyes dogfood, CI artifacts)."""
    payload: dict[str, Any] = {
        "ok": self.ok,
        "verdict": self.verdict,
        "reason": self.reason,
        "exit_code": self.exit_code,
        "trace_id": self.trace_id,
        "failed_step_indices": list(self.failed_step_indices),
        "degraded_step_indices": list(self.degraded_step_indices),
        "silent_success": self.silent_success,
        "verification": None,
    }
    if self.verification is not None:
        payload["verification"] = self.verification.to_dict()
    return payload

PrivacyScrubber

Structure-preserving PII redaction for agent traces.

Scrubs the following PII patterns from step action, observation, and result fields: - Email addresses → [EMAIL_REDACTED] - Phone numbers → [PHONE_REDACTED] - Credit card numbers → [CREDIT_CARD_REDACTED] - Social Security Numbers → [SSN_REDACTED] - IP addresses → [IP_REDACTED]

scrub(trace)

Scrub PII from a trace's step fields.

Deep-copies the trace before modification to preserve the original.

Parameters:

Name Type Description Default
trace AgentTrace

The AgentTrace to scrub.

 required

Returns:

Type Description
ScrubResult

A ScrubResult containing the scrubbed trace and replacement statistics.
Source code in src/notarize/scrubber.py 
 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
def scrub(self, trace: AgentTrace) -> ScrubResult:
    """Scrub PII from a trace's step fields.

    Deep-copies the trace before modification to preserve the original.

    Args:
        trace: The AgentTrace to scrub.

    Returns:
        A ScrubResult containing the scrubbed trace and replacement statistics.
    """
    total_replacements = 0
    matched_patterns: set[str] = set()
    new_steps: list[TraceStep] = []

    for step in trace.steps:
        scrubbed_fields: dict[str, str] = {}
        for field_name in ("action", "observation", "result", "tool_name"):
            text = getattr(step, field_name)
            if not text:
                scrubbed_fields[field_name] = text
                continue
            new_text, count, patterns = _scrub_text(text)
            scrubbed_fields[field_name] = new_text
            if count > 0:
                total_replacements += count
                matched_patterns.update(patterns)

        # Recreate the step so __post_init__ recomputes its ID from the scrubbed content.
        new_step = TraceStep(
            step_index=step.step_index,
            action=scrubbed_fields["action"],
            observation=scrubbed_fields["observation"],
            result=scrubbed_fields["result"],
            tool_name=scrubbed_fields["tool_name"],
            timestamp=step.timestamp,
        )
        new_steps.append(new_step)

    # Rebuild the AgentTrace so that the hash chain and merkle_root are recomputed.
    rebuilt = AgentTrace(
        trace_id=trace.trace_id,
        agent_name=trace.agent_name,
        task=trace.task,
        steps=new_steps,
        created_at=trace.created_at,
    )

    return ScrubResult(
        original_trace_id=trace.trace_id,
        scrubbed_trace=rebuilt,
        replacements_count=total_replacements,
        patterns_matched=sorted(matched_patterns),
    )

ScrubResult(original_trace_id, scrubbed_trace, replacements_count, patterns_matched) dataclass

Result of scrubbing PII from a trace.

Attributes:

Name Type Description
original_trace_id str

The trace_id of the original (pre-scrub) trace.
scrubbed_trace AgentTrace

A deep-copied AgentTrace with PII replaced.
replacements_count int

Total number of replacements made.
patterns_matched list[str]

List of pattern names that were triggered.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/notarize/scrubber.py 
28
29
30
31
32
33
34
35
def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "original_trace_id": self.original_trace_id,
        "scrubbed_trace": self.scrubbed_trace.to_dict(),
        "replacements_count": self.replacements_count,
        "patterns_matched": self.patterns_matched,
    }

TraceStore(path)

SQLite-backed store for traces and verification results.

All traces and results are stored in a single SQLite database. Deduplication is by trace_id for traces and by id for results.

Attributes:

Name Type Description
path

Path to the SQLite database file.
Source code in src/notarize/store.py 
42
43
44
45
46
47
48
def __init__(self, path: str | Path) -> None:
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(str(self.path))
    self._conn.row_factory = sqlite3.Row
    self._conn.executescript(self._SCHEMA)
    self._conn.commit()

close()

Close the database connection.

Source code in src/notarize/store.py 
50
51
52
def close(self) -> None:
    """Close the database connection."""
    self._conn.close()

save_trace(trace)

Store an AgentTrace (upsert by trace_id).

Parameters:

Name Type Description Default
trace AgentTrace

The AgentTrace to save.

 required
Source code in src/notarize/store.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def save_trace(self, trace: AgentTrace) -> None:
    """Store an AgentTrace (upsert by trace_id).

    Args:
        trace: The AgentTrace to save.
    """
    self._conn.execute(
        """INSERT OR REPLACE INTO traces
           (id, trace_id, agent_name, task, merkle_root, created_at, data)
           VALUES (?,?,?,?,?,?,?)""",
        (
            trace.id,
            trace.trace_id,
            trace.agent_name,
            trace.task,
            trace.merkle_root,
            trace.created_at,
            json.dumps(trace.to_dict()),
        ),
    )
    self._conn.commit()

get_trace(trace_id)

Retrieve an AgentTrace by trace_id, or None if not found.

Parameters:

Name Type Description Default
trace_id str

The user-provided trace identifier.

 required

Returns:

Type Description
AgentTrace | None

The AgentTrace, or None.
Source code in src/notarize/store.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_trace(self, trace_id: str) -> AgentTrace | None:
    """Retrieve an AgentTrace by trace_id, or None if not found.

    Args:
        trace_id: The user-provided trace identifier.

    Returns:
        The AgentTrace, or None.
    """
    row = self._conn.execute("SELECT data FROM traces WHERE trace_id=?", (trace_id,)).fetchone()
    if row is None:
        return None
    return AgentTrace.from_dict(json.loads(row["data"]))

list_traces()

Return all stored AgentTrace objects ordered by created_at.

Returns:

Type Description
list[AgentTrace]

List of AgentTrace objects, oldest first.
Source code in src/notarize/store.py 
90
91
92
93
94
95
96
97
def list_traces(self) -> list[AgentTrace]:
    """Return all stored AgentTrace objects ordered by created_at.

    Returns:
        List of AgentTrace objects, oldest first.
    """
    rows = self._conn.execute("SELECT data FROM traces ORDER BY created_at").fetchall()
    return [AgentTrace.from_dict(json.loads(r["data"])) for r in rows]

save_result(result)

Store a VerificationResult (upsert by id).

Parameters:

Name Type Description Default
result VerificationResult

The VerificationResult to save.

 required
Source code in src/notarize/store.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def save_result(self, result: VerificationResult) -> None:
    """Store a VerificationResult (upsert by id).

    Args:
        result: The VerificationResult to save.
    """
    self._conn.execute(
        """INSERT OR REPLACE INTO results
           (id, trace_id, verdict, timestamp, data)
           VALUES (?,?,?,?,?)""",
        (
            result.id,
            result.trace_id,
            result.verdict,
            result.timestamp,
            json.dumps(result.to_dict()),
        ),
    )
    self._conn.commit()

get_result(result_id)

Retrieve a VerificationResult by id, or None if not found.

Parameters:

Name Type Description Default
result_id str

The content-addressed result ID.

 required

Returns:

Type Description
VerificationResult | None

The VerificationResult, or None.
Source code in src/notarize/store.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
def get_result(self, result_id: str) -> VerificationResult | None:
    """Retrieve a VerificationResult by id, or None if not found.

    Args:
        result_id: The content-addressed result ID.

    Returns:
        The VerificationResult, or None.
    """
    row = self._conn.execute("SELECT data FROM results WHERE id=?", (result_id,)).fetchone()
    if row is None:
        return None
    return VerificationResult.from_dict(json.loads(row["data"]))

list_results()

Return all stored VerificationResult objects ordered by timestamp.

Returns:

Type Description
list[VerificationResult]

List of VerificationResult objects, oldest first.
Source code in src/notarize/store.py 
133
134
135
136
137
138
139
140
def list_results(self) -> list[VerificationResult]:
    """Return all stored VerificationResult objects ordered by timestamp.

    Returns:
        List of VerificationResult objects, oldest first.
    """
    rows = self._conn.execute("SELECT data FROM results ORDER BY timestamp").fetchall()
    return [VerificationResult.from_dict(json.loads(r["data"])) for r in rows]

AgentTrace(trace_id, agent_name, task, steps, created_at=time.time()) dataclass

A hash-chained sequence of TraceSteps with a Merkle root.

The steps form a linked chain where each step's parent_id points to the previous step's id. A Merkle root is computed from all step IDs to enable tamper detection.

Attributes:

Name Type Description
trace_id str

User-provided trace identifier.
agent_name str

Name of the agent that produced this trace.
task str

What the agent was asked to do.
steps list[TraceStep]

Ordered list of TraceStep objects.
merkle_root str

SHA-256[:16] of the sorted step IDs.
created_at float

Unix timestamp when this trace was created.
id str

SHA-256[:16] of "trace_id|agent_name|task|merkle_root".

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/notarize/trace.py 
118
119
120
121
122
123
124
125
126
127
128
def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "trace_id": self.trace_id,
        "agent_name": self.agent_name,
        "task": self.task,
        "steps": [s.to_dict() for s in self.steps],
        "merkle_root": self.merkle_root,
        "created_at": self.created_at,
        "id": self.id,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/notarize/trace.py 
130
131
132
133
134
135
136
137
138
139
140
141
@classmethod
def from_dict(cls, d: dict[str, Any]) -> AgentTrace:
    """Deserialize from a dict produced by to_dict()."""
    steps = [TraceStep.from_dict(s) for s in d.get("steps", [])]
    trace = cls(
        trace_id=d["trace_id"],
        agent_name=d["agent_name"],
        task=d["task"],
        steps=steps,
        created_at=d.get("created_at", 0.0),
    )
    return trace

TraceStep(step_index, action, observation, result, tool_name='', timestamp=time.time(), parent_id=None) dataclass

A single step in an agent execution trace.

Steps are content-addressed by their step_index, action, observation, and result. Each step points to the previous step's ID via parent_id, forming a hash chain.

Attributes:

Name Type Description
step_index int

Zero-based index of this step in the trace.
action str

What the agent did (e.g. "tool_call:search").
observation str

What the agent observed.
result str

What happened (e.g. "success", "error").
tool_name str

Optional tool name used in this step.
timestamp float

Unix timestamp of this step.
id str

SHA-256[:16] of "step_index|action|observation|result".
parent_id str | None

The previous step's ID, or None for the first step.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/notarize/trace.py 
46
47
48
49
50
51
52
53
54
55
56
57
def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "step_index": self.step_index,
        "action": self.action,
        "observation": self.observation,
        "result": self.result,
        "tool_name": self.tool_name,
        "timestamp": self.timestamp,
        "id": self.id,
        "parent_id": self.parent_id,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/notarize/trace.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_dict(cls, d: dict[str, Any]) -> TraceStep:
    """Deserialize from a dict produced by to_dict()."""
    step = cls(
        step_index=d["step_index"],
        action=d["action"],
        observation=d["observation"],
        result=d["result"],
        tool_name=d.get("tool_name", ""),
        timestamp=d.get("timestamp", 0.0),
    )
    step.parent_id = d.get("parent_id")
    return step

CompiledWorkflow(step_ids, edges, hard_edge_count, suspected_edge_count, residual_llm_count, retry_noise_count, exploration_noise_count) dataclass

Mostly deterministic workflow compiled from noisy traces.

ToolInvocation(step_id, tool, arguments=dict(), outputs=dict(), is_retry=False, is_exploration=False) dataclass

One tool step in a noisy agent trace (pre-compile).

WorkflowEdge(producer_step, consumer_step, producer_key, consumer_arg, binding, strength, evidence=(), value_fingerprint='') dataclass

Producer→consumer dependency with optional evidence (TraceCompiler).

ConsistencyVerifier

Verifies the internal consistency of an AgentTrace.

Performs the following checks: 1. Hash chain integrity: each step's id is recomputed from its content fields (step_index, action, observation, result) and compared to the stored id; also each step.parent_id == previous step.id. If either fails, hash_chain_integrity is failed and tamper_detected is added. 2. Merkle root matches recomputed value 3. Step indices are monotonically increasing from 0 4. No duplicate step IDs 5. Trace ID matches stored trace.id

verify(trace)

Verify the internal consistency of an AgentTrace.

Parameters:

Name Type Description Default
trace AgentTrace

The AgentTrace to verify.

 required

Returns:

Type Description
VerificationResult

A VerificationResult with verdict and check details.
Source code in src/notarize/verifier.py 
 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
def verify(self, trace: AgentTrace) -> VerificationResult:
    """Verify the internal consistency of an AgentTrace.

    Args:
        trace: The AgentTrace to verify.

    Returns:
        A VerificationResult with verdict and check details.
    """
    checks_passed: list[str] = []
    checks_failed: list[str] = []
    error: str | None = None

    try:
        # Check 1: Hash chain integrity - two sub-checks:
        #   (a) parent_id chain linkage
        #   (b) each step's id matches recomputed hash of its content fields
        chain_ok = True
        tamper_detected = False
        for i, step in enumerate(trace.steps):
            # (a) verify parent_id linkage
            if i == 0:
                if step.parent_id is not None:
                    chain_ok = False
            else:
                expected_parent = trace.steps[i - 1].id
                if step.parent_id != expected_parent:
                    chain_ok = False

            # (b) recompute step.id from content fields and compare
            expected_id = _sha16(
                f"{step.step_index}|{step.action}|{step.observation}|{step.result}"
            )
            if step.id != expected_id:
                chain_ok = False
                tamper_detected = True

        if chain_ok:
            checks_passed.append("hash_chain_integrity")
        else:
            checks_failed.append("hash_chain_integrity")
            if tamper_detected:
                checks_failed.append("tamper_detected")

        # Check 2: Merkle root matches recomputed value
        step_ids = sorted(s.id for s in trace.steps)
        computed_root = _sha16("|".join(step_ids)) if step_ids else _sha16("")
        if computed_root == trace.merkle_root:
            checks_passed.append("merkle_root_valid")
        else:
            checks_failed.append("merkle_root_valid")

        # Check 3: Step indices are monotonically increasing from 0
        indices_ok = True
        if trace.steps:
            if trace.steps[0].step_index != 0:
                indices_ok = False
            else:
                for i in range(1, len(trace.steps)):
                    if trace.steps[i].step_index != trace.steps[i - 1].step_index + 1:
                        indices_ok = False
                        break

        if indices_ok:
            checks_passed.append("step_indices_monotonic")
        else:
            checks_failed.append("step_indices_monotonic")

        # Check 4: No duplicate step IDs
        step_ids_list = [s.id for s in trace.steps]
        if len(step_ids_list) == len(set(step_ids_list)):
            checks_passed.append("no_duplicate_step_ids")
        else:
            checks_failed.append("no_duplicate_step_ids")

        # Check 5: Trace ID matches stored trace.id
        computed_trace_id = _sha16(
            f"{trace.trace_id}|{trace.agent_name}|{trace.task}|{trace.merkle_root}"
        )
        if computed_trace_id == trace.id:
            checks_passed.append("trace_id_valid")
        else:
            checks_failed.append("trace_id_valid")

    except Exception as exc:
        error = str(exc)
        checks_failed.append("unexpected_error")

    # Determine verdict
    if error:
        verdict = "invalid"
    elif not checks_failed:
        verdict = "verified"
    elif (
        "hash_chain_integrity" in checks_failed
        or "merkle_root_valid" in checks_failed
        or "trace_id_valid" in checks_failed
    ):
        verdict = "tampered"
    else:
        verdict = "consistent"

    return VerificationResult(
        trace_id=trace.trace_id,
        verdict=verdict,
        checks_passed=checks_passed,
        checks_failed=checks_failed,
        error=error,
        timestamp=time.time(),
    )

VerificationResult(trace_id, verdict, checks_passed, checks_failed, error, timestamp) dataclass

Result of verifying an AgentTrace.

Attributes:

Name Type Description
trace_id str

The trace_id of the verified trace.
verdict str

One of "verified", "consistent", "tampered", "invalid".
checks_passed list[str]

List of check names that passed.
checks_failed list[str]

List of check names that failed.
error str | None

Optional error message if an exception occurred.
timestamp float

Unix timestamp of the verification.
id str

Content-addressed identifier of this result.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/notarize/verifier.py 
43
44
45
46
47
48
49
50
51
52
53
def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "trace_id": self.trace_id,
        "verdict": self.verdict,
        "checks_passed": self.checks_passed,
        "checks_failed": self.checks_failed,
        "error": self.error,
        "timestamp": self.timestamp,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/notarize/verifier.py 
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def from_dict(cls, d: dict[str, Any]) -> VerificationResult:
    """Deserialize from a dict produced by to_dict()."""
    result = cls(
        trace_id=d["trace_id"],
        verdict=d["verdict"],
        checks_passed=d.get("checks_passed", []),
        checks_failed=d.get("checks_failed", []),
        error=d.get("error"),
        timestamp=d.get("timestamp", 0.0),
    )
    return result

summarize(trace)

Produce an AuditSummary for a single AgentTrace.

Parameters:

Name Type Description Default
trace AgentTrace

The AgentTrace to analyse.

 required

Returns:

Type Description
AuditSummary

An AuditSummary with risk flags and compliance score.
Source code in src/notarize/audit.py 
 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
def summarize(trace: AgentTrace) -> AuditSummary:
    """Produce an AuditSummary for a single AgentTrace.

    Args:
        trace: The AgentTrace to analyse.

    Returns:
        An AuditSummary with risk flags and compliance score.
    """
    steps = trace.steps
    total_steps = len(steps)

    # Duration
    if steps and hasattr(steps[0], "timestamp") and hasattr(steps[-1], "timestamp"):
        duration_ms = (steps[-1].timestamp - steps[0].timestamp) * 1000.0
    else:
        duration_ms = 0.0

    # Tools used (unique, in order of first appearance)
    seen: set[str] = set()
    tools_used: list[str] = []
    for step in steps:
        name = step.tool_name
        if name and name not in seen:
            seen.add(name)
            tools_used.append(name)

    # PII count - sum replacements across all text fields in all steps
    pii_fields_scrubbed = 0
    for step in steps:
        for text in (step.action, step.observation, step.result, step.tool_name):
            if text:
                _, count, _ = _scrub_text(text)
                pii_fields_scrubbed += count

    # Chain validity
    result = ConsistencyVerifier().verify(trace)
    chain_valid = result.verdict in ("verified", "consistent")

    # Risk flags
    risk_flags: list[str] = []
    if duration_ms > 300_000:
        risk_flags.append("long_duration")
    if total_steps > 50:
        risk_flags.append("many_steps")
    if pii_fields_scrubbed > 0:
        risk_flags.append("pii_detected")
    if not chain_valid:
        risk_flags.append("chain_broken")
    if not tools_used:
        risk_flags.append("no_tools_used")

    # Compliance score
    score = 100.0
    if "chain_broken" in risk_flags:
        score -= 20
    if "pii_detected" in risk_flags:
        score -= 20
    if "long_duration" in risk_flags:
        score -= 10
    if "many_steps" in risk_flags:
        score -= 10
    if "no_tools_used" in risk_flags:
        score -= 10
    compliance_score = max(0.0, min(100.0, score))

    return AuditSummary(
        session_id=trace.trace_id,
        agent_id=trace.agent_name,
        total_steps=total_steps,
        duration_ms=duration_ms,
        tools_used=tools_used,
        pii_fields_scrubbed=pii_fields_scrubbed,
        chain_valid=chain_valid,
        risk_flags=risk_flags,
        compliance_score=compliance_score,
    )

summarize_session(store, session_id)

Return AuditSummary objects for all traces belonging to a session.

A trace belongs to the session if its trace_id starts with session_id OR its agent_name equals session_id.

Parameters:

Name Type Description Default
store TraceStore

The TraceStore to query.

 required
session_id str

A session prefix or agent name to match.

 required

Returns:

Type Description
list[AuditSummary]

A list of AuditSummary objects, one per matching trace.
Source code in src/notarize/audit.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def summarize_session(store: TraceStore, session_id: str) -> list[AuditSummary]:
    """Return AuditSummary objects for all traces belonging to a session.

    A trace belongs to the session if its trace_id starts with session_id OR
    its agent_name equals session_id.

    Args:
        store: The TraceStore to query.
        session_id: A session prefix or agent name to match.

    Returns:
        A list of AuditSummary objects, one per matching trace.
    """
    all_traces = store.list_traces()
    matching = [
        t for t in all_traces if t.trace_id.startswith(session_id) or t.agent_name == session_id
    ]
    return [summarize(t) for t in matching]

assert_no_silent_success(claimed_ok, claimed_exit_code, trace, **kwargs)

Raise :class:ClosedLoopError on SILENT-SUCCESS or other gate failure.

Source code in src/notarize/closed_loop.py 
413
414
415
416
417
418
419
420
421
422
423
def assert_no_silent_success(
    claimed_ok: bool,
    claimed_exit_code: int,
    trace: AgentTrace | str | Path,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` on SILENT-SUCCESS or other gate failure."""
    outcome = gate_claimed_success(claimed_ok, claimed_exit_code, trace, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_trace_verified(trace, **kwargs)

Gate a trace and raise :class:ClosedLoopError unless outcome is ok.

Source code in src/notarize/closed_loop.py 
402
403
404
405
406
407
408
409
410
def assert_trace_verified(
    trace: AgentTrace | str | Path,
    **kwargs: Any,
) -> GateOutcome:
    """Gate a trace and raise :class:`ClosedLoopError` unless outcome is ok."""
    outcome = gate_trace(trace, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_claimed_success(claimed_ok, claimed_exit_code, trace, *, verifier=None)

Gate a process claim (exit code / success flag) against the real trace.

SILENT-SUCCESS control for assemble-style pipelines:

  • claimed_ok=True or claimed_exit_code==0 with failed/degraded steps → FAIL (exit 1), never silent pass.
  • Claim already failed → still verify empty/tamper (may be FAIL_LOUD).
  • Clean claim + clean trace → PASS.

Parameters:

Name Type Description Default
claimed_ok bool

What the process reported (success flag).

 required
claimed_exit_code int

Process exit code (0 = success claim).

 required
trace AgentTrace | str | Path

Execution trace to read.

 required
verifier ConsistencyVerifier | None

Optional consistency verifier.

 None
Source code in src/notarize/closed_loop.py 
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
def gate_claimed_success(
    claimed_ok: bool,
    claimed_exit_code: int,
    trace: AgentTrace | str | Path,
    *,
    verifier: ConsistencyVerifier | None = None,
) -> GateOutcome:
    """Gate a process claim (exit code / success flag) against the real trace.

    SILENT-SUCCESS control for assemble-style pipelines:

    * ``claimed_ok=True`` or ``claimed_exit_code==0`` with failed/degraded
      steps → FAIL (exit 1), never silent pass.
    * Claim already failed → still verify empty/tamper (may be FAIL_LOUD).
    * Clean claim + clean trace → PASS.

    Args:
        claimed_ok: What the process reported (``success`` flag).
        claimed_exit_code: Process exit code (0 = success claim).
        trace: Execution trace to read.
        verifier: Optional consistency verifier.
    """
    claim_success = bool(claimed_ok) or claimed_exit_code == 0

    # Always run integrity + degraded checks; then compare to claim.
    base = gate_trace(
        trace,
        verifier=verifier,
        refuse_degraded=True,
        refuse_failed_steps=True,
    )

    if not claim_success:
        # Process already admitted failure - surface integrity issues first.
        if base.verdict == "FAIL_LOUD":
            return base
        if base.silent_success or base.failed_step_indices or base.degraded_step_indices:
            # Consistent: claim failed and trace shows problems.
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"claimed failure (exit={claimed_exit_code}, ok={claimed_ok}) "
                    f"aligned with trace issues: {base.reason}"
                ),
                exit_code=1,
                verification=base.verification,
                trace_id=base.trace_id,
                failed_step_indices=base.failed_step_indices,
                degraded_step_indices=base.degraded_step_indices,
                silent_success=False,
            )
        if not base.ok:
            return base
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=f"claimed failure (exit={claimed_exit_code}) with clean trace",
            exit_code=1,
            verification=base.verification,
            trace_id=base.trace_id,
        )

    # Claimed success - base already fails on silent success / empty / tamper.
    if not base.ok:
        # Escalate reason if claim was success
        if base.silent_success or base.failed_step_indices or base.degraded_step_indices:
            return GateOutcome(
                ok=False,
                verdict=base.verdict if base.verdict == "FAIL_LOUD" else "FAIL",
                reason=(
                    f"SILENT-SUCCESS: claimed success (exit={claimed_exit_code}, "
                    f"ok={claimed_ok}) but {base.reason}"
                ),
                exit_code=base.exit_code if base.verdict == "FAIL_LOUD" else 1,
                verification=base.verification,
                trace_id=base.trace_id,
                failed_step_indices=base.failed_step_indices,
                degraded_step_indices=base.degraded_step_indices,
                silent_success=True,
            )
        return base

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(f"claimed success matches clean verified trace (exit={claimed_exit_code})"),
        exit_code=0,
        verification=base.verification,
        trace_id=base.trace_id,
        silent_success=False,
    )

gate_trace(trace, *, verifier=None, refuse_degraded=True, refuse_failed_steps=True)

Read one trace, verify hash-chain integrity, fail loudly on empty/wrong.

Parameters:

Name Type Description Default
trace AgentTrace | str | Path

:class:AgentTrace or path to a serialised trace.

 required
verifier ConsistencyVerifier | None

Optional verifier instance (defaults to a new one).

 None
refuse_degraded bool

If True (default), degraded/partial steps → FAIL (SILENT-SUCCESS class - clean chain must not hide soft failure).

 True
refuse_failed_steps bool

If True (default), any hard-failure step → FAIL even when the chain hashes correctly.

 True

Returns:

Type Description
GateOutcome

class:GateOutcome - callers should sys.exit(outcome.exit_code).
Source code in src/notarize/closed_loop.py 
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
def gate_trace(
    trace: AgentTrace | str | Path,
    *,
    verifier: ConsistencyVerifier | None = None,
    refuse_degraded: bool = True,
    refuse_failed_steps: bool = True,
) -> GateOutcome:
    """Read one trace, verify hash-chain integrity, fail loudly on empty/wrong.

    Args:
        trace: :class:`AgentTrace` or path to a serialised trace.
        verifier: Optional verifier instance (defaults to a new one).
        refuse_degraded: If True (default), degraded/partial steps → FAIL
            (SILENT-SUCCESS class - clean chain must not hide soft failure).
        refuse_failed_steps: If True (default), any hard-failure step → FAIL
            even when the chain hashes correctly.

    Returns:
        :class:`GateOutcome` - callers should ``sys.exit(outcome.exit_code)``.
    """
    try:
        t = _load_trace(trace)
    except ClosedLoopError as exc:
        return _fail_loud(str(exc))
    except Exception as exc:
        return _fail_loud(f"trace load failed: {exc.__class__.__name__}: {exc}")

    tid = getattr(t, "trace_id", None) or getattr(t, "id", None)

    steps = getattr(t, "steps", None)
    if steps is None:
        return _fail_loud("trace has no steps attribute", tid)
    if len(steps) == 0:
        return _fail_loud("empty trace - write-only empty log is ornament", tid)

    failed_ix = failed_step_indices(t)
    degraded_ix = degraded_step_indices(t)

    v = verifier or ConsistencyVerifier()
    try:
        result = v.verify(t)
    except Exception as exc:
        return _fail_loud(f"verify raised: {exc.__class__.__name__}: {exc}", tid)

    if result.verdict not in {"verified", "consistent"}:
        return _fail(
            f"verdict={result.verdict} failed={result.checks_failed} error={result.error!r}",
            tid,
            verification=result,
            failed=failed_ix,
            degraded=degraded_ix,
        )

    # Chain is intact - still refuse failed / degraded steps (SILENT-SUCCESS).
    if refuse_failed_steps and failed_ix:
        return _fail(
            f"SILENT-SUCCESS: chain verified but failed steps at {failed_ix} "
            f"- refuse success (assemble/exit-0 degraded class)",
            tid,
            verification=result,
            failed=failed_ix,
            degraded=degraded_ix,
            silent_success=True,
        )

    if refuse_degraded and degraded_ix:
        return _fail(
            f"SILENT-SUCCESS: chain verified but degraded steps at {degraded_ix} "
            f"- refuse clean PASS",
            tid,
            verification=result,
            failed=failed_ix,
            degraded=degraded_ix,
            silent_success=True,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"verdict={result.verdict} checks_passed={len(result.checks_passed)}",
        exit_code=0,
        verification=result,
        trace_id=tid,
        failed_step_indices=tuple(failed_ix),
        degraded_step_indices=tuple(degraded_ix),
        silent_success=False,
    )

step_is_degraded(step)

True when the step is partial/degraded (not a clean success).

Source code in src/notarize/closed_loop.py 
133
134
135
136
137
138
139
140
141
142
143
144
def step_is_degraded(step: TraceStep) -> bool:
    """True when the step is partial/degraded (not a clean success)."""
    if step_is_failed(step):
        return False
    r = _norm_result(step.result)
    if r in _DEGRADED_RESULTS:
        return True
    head = r.split(":", 1)[0]
    if head in _DEGRADED_RESULTS:
        return True
    blob = f"{step.result} {step.observation} {step.action}"
    return bool(_DEGRADED_TEXT.search(blob))

step_is_failed(step)

True when the step records a hard failure result.

Source code in src/notarize/closed_loop.py 
123
124
125
126
127
128
129
130
def step_is_failed(step: TraceStep) -> bool:
    """True when the step records a hard failure result."""
    r = _norm_result(step.result)
    if r in _FAILURE_RESULTS:
        return True
    # "error:timeout", "failed:disk"
    head = r.split(":", 1)[0]
    return head in _FAILURE_RESULTS

compare_traces(baseline, candidate)

Compare two AgentTraces step by step.

For each step position both traces have in common, compute the similarity between the concatenated "action|observation|result" strings. Steps only present in one trace are marked "added" or "removed" with similarity 0.

Parameters:

Name Type Description Default
baseline AgentTrace

The reference AgentTrace.

 required
candidate AgentTrace

The AgentTrace to compare against the baseline.

 required

Returns:

Type Description
TraceComparison

A TraceComparison with per-step breakdowns and an overall verdict.
Source code in src/notarize/compare.py 
 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
def compare_traces(baseline: AgentTrace, candidate: AgentTrace) -> TraceComparison:
    """Compare two AgentTraces step by step.

    For each step position both traces have in common, compute the similarity
    between the concatenated "action|observation|result" strings.  Steps only
    present in one trace are marked "added" or "removed" with similarity 0.

    Args:
        baseline: The reference AgentTrace.
        candidate: The AgentTrace to compare against the baseline.

    Returns:
        A TraceComparison with per-step breakdowns and an overall verdict.
    """
    baseline_steps = baseline.steps
    candidate_steps = candidate.steps

    # Handle the empty-traces edge case early.
    if not baseline_steps and not candidate_steps:
        return TraceComparison(
            baseline_id=baseline.trace_id,
            candidate_id=candidate.trace_id,
            step_comparisons=[],
            first_divergence=None,
            similarity_score=1.0,
            verdict="identical",
        )

    max_len = max(len(baseline_steps), len(candidate_steps))
    comparisons: list[StepComparison] = []

    for i in range(max_len):
        has_baseline = i < len(baseline_steps)
        has_candidate = i < len(candidate_steps)

        if has_baseline and has_candidate:
            b_text = _step_text(baseline_steps[i])
            c_text = _step_text(candidate_steps[i])
            sim = _similarity(b_text, c_text)
            status = "match" if sim >= 0.95 else "changed"
            comparisons.append(
                StepComparison(
                    step_index=i,
                    status=status,
                    baseline_action=baseline_steps[i].action,
                    candidate_action=candidate_steps[i].action,
                    similarity=sim,
                )
            )
        elif has_baseline:
            comparisons.append(
                StepComparison(
                    step_index=i,
                    status="removed",
                    baseline_action=baseline_steps[i].action,
                    candidate_action=None,
                    similarity=0.0,
                )
            )
        else:
            comparisons.append(
                StepComparison(
                    step_index=i,
                    status="added",
                    baseline_action=None,
                    candidate_action=candidate_steps[i].action,
                    similarity=0.0,
                )
            )

    first_divergence: int | None = None
    for sc in comparisons:
        if sc.status != "match":
            first_divergence = sc.step_index
            break

    total_sim = sum(sc.similarity for sc in comparisons)
    similarity_score = total_sim / len(comparisons)

    if similarity_score >= 0.95:
        verdict = "identical"
    elif similarity_score >= 0.6:
        verdict = "minor_drift"
    else:
        verdict = "major_divergence"

    return TraceComparison(
        baseline_id=baseline.trace_id,
        candidate_id=candidate.trace_id,
        step_comparisons=comparisons,
        first_divergence=first_divergence,
        similarity_score=similarity_score,
        verdict=verdict,
    )

to_compliance_report(trace, standard='SOC2')

Generate a formal compliance report in markdown.

Parameters:

Name Type Description Default
trace AgentTrace

The AgentTrace to report on.

 required
standard str

One of 'SOC2', 'HIPAA', 'GDPR'.

 'SOC2'

Returns:

Type Description
str

A markdown-formatted compliance report string.

Raises:

Type Description
ValueError

If an unknown standard is specified.
Source code in src/notarize/timeline.py 
 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
def to_compliance_report(trace: AgentTrace, standard: str = "SOC2") -> str:
    """Generate a formal compliance report in markdown.

    Args:
        trace: The AgentTrace to report on.
        standard: One of 'SOC2', 'HIPAA', 'GDPR'.

    Returns:
        A markdown-formatted compliance report string.

    Raises:
        ValueError: If an unknown standard is specified.
    """
    if standard not in _KNOWN_STANDARDS:
        raise ValueError(
            f"Unknown compliance standard {standard!r}. Choose from: {sorted(_KNOWN_STANDARDS)}"
        )

    created_at_iso = datetime.datetime.fromtimestamp(
        trace.created_at, tz=datetime.timezone.utc
    ).isoformat()
    generated_at = datetime.datetime.now(tz=datetime.timezone.utc).isoformat()

    lines: list[str] = []

    # Title
    lines.append(f"# {standard} Compliance Report - Trace {trace.trace_id}")
    lines.append("")

    # Metadata
    lines.append("## Metadata")
    lines.append("")
    lines.append("| Field | Value |")
    lines.append("|---|---|")
    lines.append(f"| Agent | {trace.agent_name} |")
    lines.append(f"| Task | {trace.task} |")
    lines.append(f"| Steps | {len(trace.steps)} |")
    lines.append(f"| Created At | {created_at_iso} |")
    lines.append(f"| Merkle Root | `{trace.merkle_root}` |")
    lines.append(f"| Trace ID | `{trace.trace_id}` |")
    lines.append("")

    # Standards-specific section
    if standard == "SOC2":
        lines.append("## SOC2 Trust Service Criteria")
        lines.append("")
        lines.append("### Availability")
        lines.append("")
        lines.append(f"- Trace recorded {len(trace.steps)} execution steps.")
        lines.append("- All steps are persisted and available for audit retrieval.")
        lines.append("")
        lines.append("### Confidentiality")
        lines.append("")
        lines.append(
            "- Trace data should be scrubbed of PII before storage using `notarize scrub`."
        )
        lines.append("- Access to the trace store should be restricted to authorised principals.")
        lines.append("")
        lines.append("### Processing Integrity")
        lines.append("")
        lines.append("- Chain integrity is enforced via the Merkle root of sorted step IDs.")
        lines.append(f"- Current Merkle root: `{trace.merkle_root}`.")
        lines.append("- Run `notarize verify` to confirm chain has not been tampered with.")

    elif standard == "HIPAA":
        lines.append("## HIPAA Compliance Notes")
        lines.append("")
        lines.append("### PHI Handling")
        lines.append("")
        lines.append("- Protected Health Information (PHI) must not appear in trace step fields.")
        lines.append("- Apply `notarize scrub` before storing or transmitting traces.")
        lines.append("")
        lines.append("### Access Controls")
        lines.append("")
        lines.append(
            "- The trace store (SQLite database) must be protected with file-system permissions."
        )
        lines.append("- Only authorised personnel should have read access to stored traces.")
        lines.append("")
        lines.append("### Audit Trail Completeness")
        lines.append("")
        lines.append(
            f"- This trace contains {len(trace.steps)} step(s), providing a complete audit trail."
        )
        lines.append(
            "- Merkle root verification ensures no steps have been added, removed, or altered."
        )
        lines.append(f"- Merkle root: `{trace.merkle_root}`.")

    elif standard == "GDPR":
        lines.append("## GDPR Compliance Notes")
        lines.append("")
        lines.append("### Data Minimisation")
        lines.append("")
        lines.append("- Traces should capture only the minimum data necessary for auditability.")
        lines.append("- Use `notarize scrub` to remove personal data before retention.")
        lines.append("")
        lines.append("### Purpose Limitation")
        lines.append("")
        lines.append(
            "- Traces are collected solely for agent execution auditability"
            " and compliance purposes."
        )
        lines.append("- Data must not be repurposed for unrelated processing activities.")
        lines.append("")
        lines.append("### Retention")
        lines.append("")
        lines.append("- Define and enforce a retention policy for stored traces.")
        lines.append("- Purge traces that are no longer required for their original purpose.")

    lines.append("")

    # Step table
    lines.append("## Step Summary")
    lines.append("")
    lines.append("| Step | Action | Tool | Result |")
    lines.append("|---|---|---|---|")
    for step in trace.steps:
        tool = step.tool_name or "-"
        lines.append(f"| {step.step_index} | {step.action} | {tool} | {step.result} |")
    lines.append("")

    # Footer
    lines.append("---")
    lines.append("")
    lines.append(f"*Generated at {generated_at} by notarize.*")

    return "\n".join(lines)

to_csv(trace)

Export trace as CSV: step_index,action,input_summary,output_summary,duration_ms,timestamp

Source code in src/notarize/timeline.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def to_csv(trace: AgentTrace) -> str:
    """Export trace as CSV: step_index,action,input_summary,output_summary,duration_ms,timestamp"""
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(
        ["step_index", "action", "input_summary", "output_summary", "duration_ms", "timestamp"]
    )

    steps = trace.steps
    for i, step in enumerate(steps):
        duration_ms = 0.0 if i == 0 else (step.timestamp - steps[i - 1].timestamp) * 1000.0

        input_summary = (step.observation or "")[:80]
        output_summary = (step.result or "")[:80]
        ts = datetime.datetime.fromtimestamp(step.timestamp, tz=datetime.timezone.utc).isoformat()

        writer.writerow(
            [step.step_index, step.action, input_summary, output_summary, duration_ms, ts]
        )

    return buf.getvalue()

to_timeline_json(trace)

Export as JSON array suitable for timeline visualizations.

Source code in src/notarize/timeline.py 
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
def to_timeline_json(trace: AgentTrace) -> str:
    """Export as JSON array suitable for timeline visualizations."""
    steps = trace.steps
    result = []

    for i, step in enumerate(steps):
        if i < len(steps) - 1:
            duration_ms = (steps[i + 1].timestamp - step.timestamp) * 1000.0
        else:
            duration_ms = 0.0

        start_time = datetime.datetime.fromtimestamp(
            step.timestamp, tz=datetime.timezone.utc
        ).isoformat()

        result.append(
            {
                "step_index": step.step_index,
                "action": step.action,
                "tool_name": step.tool_name,
                "start_time": start_time,
                "duration_ms": duration_ms,
                "status": step.result,
                "id": step.id,
            }
        )

    return json.dumps(result, indent=2)

assert_compiled_workflow_ok(workflow=None, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_compiled_workflow is ok.

Source code in src/notarize/trace_compile.py 
516
517
518
519
520
521
522
523
524
def assert_compiled_workflow_ok(
    workflow: CompiledWorkflow | Sequence[WorkflowEdge] | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_compiled_workflow` is ok."""
    outcome = gate_compiled_workflow(workflow, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

compile_trace_workflow(invocations, *, drop_retries=True, drop_exploration=True)

Mine producer→consumer edges from tool invocations (deterministic).

Hard edge rule (TraceCompiler): Admit a hard edge only when a consumer argument value is uniquely attributable to exactly one earlier producer's output value. Evidence tuple = (producer_step, producer_key, consumer_arg, fingerprint).

Ambiguous (value appears in 0 or ≥2 producers) → suspected edge with empty/weak evidence and no hard ordering obligation.

Argument values that match no producer
  • empty / None → constant (if literal-looking) or user_input
  • otherwise → llm_residual suspected edge from previous step (weak)
Source code in src/notarize/trace_compile.py 
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
def compile_trace_workflow(
    invocations: Sequence[ToolInvocation | dict[str, Any]],
    *,
    drop_retries: bool = True,
    drop_exploration: bool = True,
) -> CompiledWorkflow:
    """Mine producer→consumer edges from tool invocations (deterministic).

    Hard edge rule (TraceCompiler):
      Admit a **hard** edge only when a consumer argument value is uniquely
      attributable to exactly one earlier producer's output value. Evidence
      tuple = (producer_step, producer_key, consumer_arg, fingerprint).

    Ambiguous (value appears in 0 or ≥2 producers) → **suspected** edge with
    empty/weak evidence and **no hard ordering** obligation.

    Argument values that match no producer:
      * empty / None → constant (if literal-looking) or user_input
      * otherwise → llm_residual suspected edge from previous step (weak)
    """
    invs = [_as_invocation(x) for x in invocations]
    retry_n = sum(1 for i in invs if i.is_retry)
    explor_n = sum(1 for i in invs if i.is_exploration)

    kept: list[ToolInvocation] = []
    for inv in invs:
        if drop_retries and inv.is_retry:
            continue
        if drop_exploration and inv.is_exploration:
            continue
        kept.append(inv)

    # Map fingerprint → list of (step_id, out_key) producers
    producers_by_fp: dict[str, list[tuple[str, str]]] = {}
    for inv in kept:
        for key, val in inv.outputs.items():
            fp = _fp(val)
            if not fp:
                continue
            producers_by_fp.setdefault(fp, []).append((inv.step_id, key))

    edges: list[WorkflowEdge] = []
    residual = 0
    step_ids = tuple(i.step_id for i in kept)

    for inv in kept:
        for arg_name, arg_val in inv.arguments.items():
            fp = _fp(arg_val)
            if not fp:
                # empty arg - treat as user_input residual (no hard edge)
                residual += 1
                edges.append(
                    WorkflowEdge(
                        producer_step="",
                        consumer_step=inv.step_id,
                        producer_key="",
                        consumer_arg=arg_name,
                        binding="user_input",
                        strength="suspected",
                        evidence=(),
                        value_fingerprint="",
                    )
                )
                continue

            matches = producers_by_fp.get(fp, [])
            # only earlier producers
            earlier = [
                (sid, key)
                for sid, key in matches
                if sid != inv.step_id and step_ids.index(sid) < step_ids.index(inv.step_id)
            ]

            if len(earlier) == 1:
                prod_step, prod_key = earlier[0]
                evidence = (
                    f"producer={prod_step}",
                    f"out={prod_key}",
                    f"arg={arg_name}",
                    f"fp={fp[:64]}",
                )
                edges.append(
                    WorkflowEdge(
                        producer_step=prod_step,
                        consumer_step=inv.step_id,
                        producer_key=prod_key,
                        consumer_arg=arg_name,
                        binding="copied_output",
                        strength="hard",
                        evidence=evidence,
                        value_fingerprint=fp[:128],
                    )
                )
            elif len(earlier) > 1:
                # ambiguous - suspected, no hard ordering
                prod_step, prod_key = earlier[0]
                edges.append(
                    WorkflowEdge(
                        producer_step=prod_step,
                        consumer_step=inv.step_id,
                        producer_key=prod_key,
                        consumer_arg=arg_name,
                        binding="copied_output",
                        strength="suspected",
                        evidence=(f"ambiguous_producers={len(earlier)}",),
                        value_fingerprint=fp[:128],
                    )
                )
            else:
                # no producer - constant if short literal-like, else llm residual
                is_const = isinstance(arg_val, (int, float, bool)) or (
                    isinstance(arg_val, str) and len(arg_val) < 40 and " " not in arg_val.strip()
                )
                if is_const:
                    edges.append(
                        WorkflowEdge(
                            producer_step="",
                            consumer_step=inv.step_id,
                            producer_key="",
                            consumer_arg=arg_name,
                            binding="constant",
                            strength="suspected",
                            evidence=(f"literal={fp[:40]}",),
                            value_fingerprint=fp[:128],
                        )
                    )
                else:
                    residual += 1
                    prev = ""
                    idx = step_ids.index(inv.step_id)
                    if idx > 0:
                        prev = step_ids[idx - 1]
                    edges.append(
                        WorkflowEdge(
                            producer_step=prev,
                            consumer_step=inv.step_id,
                            producer_key="",
                            consumer_arg=arg_name,
                            binding="llm_residual",
                            strength="suspected",
                            evidence=(),
                            value_fingerprint=fp[:128],
                        )
                    )

    hard_n = sum(1 for e in edges if e.strength == "hard")
    sus_n = sum(1 for e in edges if e.strength == "suspected")
    return CompiledWorkflow(
        step_ids=step_ids,
        edges=tuple(edges),
        hard_edge_count=hard_n,
        suspected_edge_count=sus_n,
        residual_llm_count=residual,
        retry_noise_count=retry_n,
        exploration_noise_count=explor_n,
    )

gate_compiled_workflow(workflow=None, *, invocations=None, require_workflow=True, require_hard_edges=False, min_hard_edges=0, refuse_hard_without_evidence=True, refuse_all_llm_residual=True, max_residual_ratio=1.0)

Refuse unattested or purely residual compiled workflows (TRACE-COMPILE).

Rules:

  • No workflow when required → FAIL_LOUD
  • Hard edge without evidence → FAIL_LOUD (audit break)
  • require_hard_edges and hard_edge_count < min → FAIL
  • All bindings residual LLM when refuse_all_llm_residual and edges exist → FAIL
  • residual ratio > max_residual_ratio → FAIL
  • Suspected edges alone do not fail ordering (TraceCompiler)
  • Clean hard edges with evidence → PASS
Source code in src/notarize/trace_compile.py 
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
def gate_compiled_workflow(
    workflow: CompiledWorkflow | Sequence[WorkflowEdge] | None = None,
    *,
    invocations: Sequence[ToolInvocation | dict[str, Any]] | None = None,
    require_workflow: bool = True,
    require_hard_edges: bool = False,
    min_hard_edges: int = 0,
    refuse_hard_without_evidence: bool = True,
    refuse_all_llm_residual: bool = True,
    max_residual_ratio: float = 1.0,
) -> GateOutcome:
    """Refuse unattested or purely residual compiled workflows (TRACE-COMPILE).

    Rules:

    * No workflow when required → **FAIL_LOUD**
    * Hard edge without evidence → **FAIL_LOUD** (audit break)
    * ``require_hard_edges`` and hard_edge_count < min → **FAIL**
    * All bindings residual LLM when refuse_all_llm_residual and edges exist → **FAIL**
    * residual ratio > max_residual_ratio → **FAIL**
    * Suspected edges alone do **not** fail ordering (TraceCompiler)
    * Clean hard edges with evidence → **PASS**
    """
    wf: CompiledWorkflow | None = None
    edges: list[WorkflowEdge] = []

    if invocations is not None and workflow is None:
        wf = compile_trace_workflow(invocations)
        edges = list(wf.edges)
    elif isinstance(workflow, CompiledWorkflow):
        wf = workflow
        edges = list(workflow.edges)
    elif workflow is not None:
        try:
            edges = [_as_edge(e) for e in workflow]
        except (TypeError, ValueError) as exc:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=f"TRACE-COMPILE: invalid edge payload: {exc}",
                exit_code=2,
            )
        hard_n = sum(1 for e in edges if e.strength == "hard")
        sus_n = len(edges) - hard_n
        residual = sum(1 for e in edges if e.binding == "llm_residual")
        wf = CompiledWorkflow(
            step_ids=tuple(
                dict.fromkeys(
                    [e.producer_step for e in edges if e.producer_step]
                    + [e.consumer_step for e in edges if e.consumer_step]
                )
            ),
            edges=tuple(edges),
            hard_edge_count=hard_n,
            suspected_edge_count=sus_n,
            residual_llm_count=residual,
            retry_noise_count=0,
            exploration_noise_count=0,
        )
    else:
        edges = []
        wf = None

    if require_workflow and (wf is None or (not edges and not (wf and wf.step_ids))):
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                "TRACE-COMPILE: no compiled workflow - cannot promote empty "
                "trace mining as a skill/workflow (arXiv 2608.02680)"
            ),
            exit_code=2,
        )

    if wf is None:
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="TRACE-COMPILE: no workflow required; nothing to gate",
            exit_code=0,
        )

    if refuse_hard_without_evidence:
        bad = hard_edges_missing_evidence(wf.edges)
        if bad:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    f"TRACE-COMPILE: {len(bad)} hard edge(s) lack auditable evidence "
                    f"(producer/consumer/fingerprint) - refuse unattested ordering "
                    f"ids={[f'{b.producer_step}->{b.consumer_step}' for b in bad[:6]]}"
                ),
                exit_code=2,
            )

    if require_hard_edges and wf.hard_edge_count < max(min_hard_edges, 1):
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"TRACE-COMPILE: hard_edge_count={wf.hard_edge_count} < "
                f"required={max(min_hard_edges, 1)} - workflow has no unique "
                "producer→consumer attributions (mostly noise/retries)"
            ),
            exit_code=1,
        )

    if min_hard_edges > 0 and wf.hard_edge_count < min_hard_edges:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"TRACE-COMPILE: hard_edge_count={wf.hard_edge_count} < "
                f"min_hard_edges={min_hard_edges}"
            ),
            exit_code=1,
        )

    edge_n = len(wf.edges)
    if edge_n > 0 and refuse_all_llm_residual:
        residual = sum(1 for e in wf.edges if e.binding == "llm_residual")
        if residual == edge_n:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    "TRACE-COMPILE: all edges are llm_residual - compiled workflow "
                    "is not mostly-deterministic; refuse promotion to replay skill "
                    "(TraceCompiler residual class)"
                ),
                exit_code=1,
            )
        ratio = residual / edge_n
        if ratio > max_residual_ratio:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"TRACE-COMPILE: residual_llm ratio={ratio:.2f} > "
                    f"max={max_residual_ratio:.2f} (residual={residual}/{edge_n})"
                ),
                exit_code=1,
            )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"TRACE-COMPILE ok: steps={len(wf.step_ids)} hard={wf.hard_edge_count} "
            f"suspected={wf.suspected_edge_count} residual={wf.residual_llm_count} "
            f"retries_dropped={wf.retry_noise_count}"
        ),
        exit_code=0,
    )

hard_edges_missing_evidence(edges)

Hard edges must carry non-empty evidence tuples.

Source code in src/notarize/trace_compile.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def hard_edges_missing_evidence(edges: Sequence[WorkflowEdge]) -> list[WorkflowEdge]:
    """Hard edges must carry non-empty evidence tuples."""
    bad: list[WorkflowEdge] = []
    for e in edges:
        if e.strength != "hard":
            continue
        if (
            not e.evidence
            or not e.producer_step
            or not e.consumer_step
            or (not e.value_fingerprint and e.binding == "copied_output")
        ):
            bad.append(e)
    return bad