Skip to content

API reference

Core — envelope, model, gate, registry, harness

halo_agent_harness.core.model

Core domain model — the agent I/O envelope and the two-axis authority/decision model.

Implements harness-protocol.md §2 (envelope) and §3 (authority + decision). This module is framework-free and dependency-free (ADR-0005, INV-5).

AuthorityLevel

Bases: IntEnum

An agent's static capability ceiling (spec §3.1). Ordered: higher = more authority.

MUST NOT change at runtime (no self-escalation, INV-3).

Source code in src/halo_agent_harness/core/model.py 
15
16
17
18
19
20
21
22
23
24
25
class AuthorityLevel(enum.IntEnum):
    """An agent's static capability ceiling (spec §3.1). Ordered: higher = more authority.

    MUST NOT change at runtime (no self-escalation, INV-3).
    """

    OBSERVE = 1
    SUGGEST = 2
    ALERT = 3
    RATE_LIMIT = 4
    BLOCK = 5

DecisionAction

Bases: Enum

The dynamic outcome of a single invocation (spec §3.2).

Source code in src/halo_agent_harness/core/model.py 
28
29
30
31
32
33
34
35
class DecisionAction(enum.Enum):
    """The dynamic outcome of a single invocation (spec §3.2)."""

    ALLOW = "ALLOW"
    BLOCK = "BLOCK"
    ALERT = "ALERT"
    SUGGEST = "SUGGEST"
    DEFER = "DEFER"

AgentInput dataclass

One invocation's input (spec §2.1). tenantId/userId are the multi-tenancy boundary.

Source code in src/halo_agent_harness/core/model.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@dataclass(frozen=True)
class AgentInput:
    """One invocation's input (spec §2.1). ``tenantId``/``userId`` are the multi-tenancy boundary."""

    tenant_id: str
    user_id: str
    context: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)

    def is_scoped(self) -> bool:
        """True if tenant and user scope are both present and non-empty (spec §2.1)."""
        return bool(self.tenant_id) and bool(self.user_id)

    @property
    def correlation_id(self) -> str | None:
        """Correlation/trace id to propagate into every port call (spec §7.5)."""
        cid = self.metadata.get("correlationId") or self.metadata.get("correlation_id")
        return str(cid) if cid is not None else None

correlation_id property 

correlation_id

Correlation/trace id to propagate into every port call (spec §7.5).

is_scoped 

is_scoped()

True if tenant and user scope are both present and non-empty (spec §2.1).

Source code in src/halo_agent_harness/core/model.py 
90
91
92
def is_scoped(self) -> bool:
    """True if tenant and user scope are both present and non-empty (spec §2.1)."""
    return bool(self.tenant_id) and bool(self.user_id)

Decision dataclass

The outcome an agent proposes; auto_enforced is owned by the harness (spec §2.2, §4).

An agent MUST NOT set auto_enforced; the harness overwrites it via the confidence gate.

Source code in src/halo_agent_harness/core/model.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@dataclass
class Decision:
    """The outcome an agent proposes; ``auto_enforced`` is owned by the harness (spec §2.2, §4).

    An agent MUST NOT set ``auto_enforced``; the harness overwrites it via the confidence gate.
    """

    action: DecisionAction
    confidence: float
    rationale: str
    auto_enforced: bool = False

    def validate(self) -> None:
        """Protocol-level validation of the decision shape (spec §2.2)."""
        if not (0.0 <= self.confidence <= 1.0):
            raise ValueError(f"confidence {self.confidence!r} out of range [0.0, 1.0]")
        if self.action in (DecisionAction.BLOCK, DecisionAction.ALERT) and not self.rationale.strip():
            raise ValueError(f"{self.action.value} decisions MUST carry a human-readable rationale")

validate 

validate()

Protocol-level validation of the decision shape (spec §2.2).

Source code in src/halo_agent_harness/core/model.py 
113
114
115
116
117
118
def validate(self) -> None:
    """Protocol-level validation of the decision shape (spec §2.2)."""
    if not (0.0 <= self.confidence <= 1.0):
        raise ValueError(f"confidence {self.confidence!r} out of range [0.0, 1.0]")
    if self.action in (DecisionAction.BLOCK, DecisionAction.ALERT) and not self.rationale.strip():
        raise ValueError(f"{self.action.value} decisions MUST carry a human-readable rationale")

AgentOutput dataclass

One invocation's output envelope (spec §2.2).

Source code in src/halo_agent_harness/core/model.py 
121
122
123
124
125
126
127
128
129
130
131
@dataclass(frozen=True)
class AgentOutput:
    """One invocation's output envelope (spec §2.2)."""

    decision: Decision
    agent_name: str
    executed_at: datetime

    @staticmethod
    def now(decision: Decision, agent_name: str) -> "AgentOutput":
        return AgentOutput(decision=decision, agent_name=agent_name, executed_at=datetime.now(timezone.utc))

min_authority_for 

min_authority_for(action)

The minimum authority level required to emit action (spec §3.3).

Source code in src/halo_agent_harness/core/model.py 
59
60
61
def min_authority_for(action: DecisionAction) -> AuthorityLevel:
    """The minimum authority level required to emit ``action`` (spec §3.3)."""
    return _ACTION_MIN_AUTHORITY[action]

action_within_authority 

action_within_authority(action, authority)

True if an agent at authority is permitted to emit action (spec §3.3).

Source code in src/halo_agent_harness/core/model.py 
64
65
66
def action_within_authority(action: DecisionAction, authority: AuthorityLevel) -> bool:
    """True if an agent at ``authority`` is permitted to emit ``action`` (spec §3.3)."""
    return authority >= _ACTION_MIN_AUTHORITY[action]

action_precedence 

action_precedence(action)

The Decision-Hierarchy precedence of an action (§3.3): higher = safer/stricter.

Source code in src/halo_agent_harness/core/model.py 
69
70
71
def action_precedence(action: DecisionAction) -> int:
    """The Decision-Hierarchy precedence of an action (§3.3): higher = safer/stricter."""
    return _ACTION_PRECEDENCE[action]

reconcile 

reconcile(actions)

Reconcile competing decisions toward the safer action (spec §3.3, §6.2).

Source code in src/halo_agent_harness/core/model.py 
74
75
76
77
78
def reconcile(actions: list[DecisionAction]) -> DecisionAction:
    """Reconcile competing decisions toward the safer action (spec §3.3, §6.2)."""
    if not actions:
        raise ValueError("cannot reconcile an empty list of actions")
    return max(actions, key=lambda a: _ACTION_PRECEDENCE[a])

halo_agent_harness.core.gate

The confidence gate (spec §4, ADR-0003).

Centralized, runs on every invocation, and cannot be disabled. It is the only place auto_enforced is set. Agents cannot reach it.

ConfidenceGate

Decides auto_enforced for a decision. There is no 'disabled' state (G-4).

The gate never trusts an incoming auto_enforced flag — it always recomputes it.

Source code in src/halo_agent_harness/core/gate.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class ConfidenceGate:
    """Decides ``auto_enforced`` for a decision. There is no 'disabled' state (G-4).

    The gate never trusts an incoming ``auto_enforced`` flag — it always recomputes it.
    """

    def threshold_for(self, authority: AuthorityLevel) -> float:
        """Effective auto-enforce threshold for an authority level (never below BASE_THRESHOLD)."""
        return max(BASE_THRESHOLD, _THRESHOLDS.get(authority, BASE_THRESHOLD))

    def evaluate(self, decision: Decision, authority: AuthorityLevel) -> bool:
        """Compute whether ``decision`` may auto-enforce, and stamp it onto the decision.

        Returns the resulting ``auto_enforced`` value (also written to ``decision.auto_enforced``).
        """
        auto = self._compute(decision, authority)
        decision.auto_enforced = auto
        return auto

    def _compute(self, decision: Decision, authority: AuthorityLevel) -> bool:
        # G-5: OBSERVE/SUGGEST-authority agents never auto-enforce, regardless of confidence.
        if authority <= AuthorityLevel.SUGGEST:
            return False
        # DEFER/SUGGEST outcomes always route to a human (spec §3.3).
        if decision.action in _ALWAYS_REVIEW_ACTIONS:
            return False
        # ALLOW has no external effect; it is "enforced" trivially only when confidence clears the bar.
        return decision.confidence >= self.threshold_for(authority)

    def is_bypass(self, decision: Decision, authority: AuthorityLevel) -> bool:
        """A bypass is an auto-enforced decision that should not have cleared the gate.

        In a correct system this is always False (spec §4.2: ``confidence_gate_bypass_total`` == 0).
        Used defensively by the harness to feed the bypass counter.
        """
        if not decision.auto_enforced:
            return False
        return self._compute(decision, authority) is False

threshold_for 

threshold_for(authority)

Effective auto-enforce threshold for an authority level (never below BASE_THRESHOLD).

Source code in src/halo_agent_harness/core/gate.py 
31
32
33
def threshold_for(self, authority: AuthorityLevel) -> float:
    """Effective auto-enforce threshold for an authority level (never below BASE_THRESHOLD)."""
    return max(BASE_THRESHOLD, _THRESHOLDS.get(authority, BASE_THRESHOLD))

evaluate 

evaluate(decision, authority)

Compute whether decision may auto-enforce, and stamp it onto the decision.

Returns the resulting auto_enforced value (also written to decision.auto_enforced).

Source code in src/halo_agent_harness/core/gate.py 
35
36
37
38
39
40
41
42
def evaluate(self, decision: Decision, authority: AuthorityLevel) -> bool:
    """Compute whether ``decision`` may auto-enforce, and stamp it onto the decision.

    Returns the resulting ``auto_enforced`` value (also written to ``decision.auto_enforced``).
    """
    auto = self._compute(decision, authority)
    decision.auto_enforced = auto
    return auto

is_bypass 

is_bypass(decision, authority)

A bypass is an auto-enforced decision that should not have cleared the gate.

In a correct system this is always False (spec §4.2: confidence_gate_bypass_total == 0). Used defensively by the harness to feed the bypass counter.

Source code in src/halo_agent_harness/core/gate.py 
54
55
56
57
58
59
60
61
62
def is_bypass(self, decision: Decision, authority: AuthorityLevel) -> bool:
    """A bypass is an auto-enforced decision that should not have cleared the gate.

    In a correct system this is always False (spec §4.2: ``confidence_gate_bypass_total`` == 0).
    Used defensively by the harness to feed the bypass counter.
    """
    if not decision.auto_enforced:
        return False
    return self._compute(decision, authority) is False

halo_agent_harness.core.registry

The tool registry — the governance boundary for agent capabilities (spec §5, ADR-0004).

Default-deny: a tool not in an agent's allowlist is unreachable. No wildcards. Registration of a wildcard allowlist is rejected. The registry also holds the concrete tool implementations.

ToolRegistry

Holds tool implementations and per-agent allowlists; authorizes every call.

Source code in src/halo_agent_harness/core/registry.py 
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
class ToolRegistry:
    """Holds tool implementations and per-agent allowlists; authorizes every call."""

    def __init__(self) -> None:
        self._tools: dict[str, RegisteredTool] = {}
        self._allowlists: dict[str, frozenset[str]] = {}

    # -- registration ----------------------------------------------------
    def register_tool(
        self,
        name: str,
        impl: ToolImpl,
        *,
        description: str = "",
        parameters: dict | None = None,
        side_effect: str = "read",
    ) -> None:
        if "*" in name:
            raise ToolRegistrationError(f"tool name {name!r} must not contain a wildcard")
        if side_effect not in _SIDE_EFFECTS:
            raise ToolRegistrationError(f"unknown side_effect {side_effect!r}")
        self._tools[name] = RegisteredTool(
            name=name,
            description=description,
            parameters=parameters or {},
            side_effect=side_effect,
            impl=impl,
        )

    def grant(self, agent_name: str, tool_names: frozenset[str] | set[str] | list[str]) -> None:
        """Set an agent's explicit tool allowlist (spec §5 T-1, T-3). Wildcards are rejected."""
        names = frozenset(tool_names)
        for n in names:
            if "*" in n:
                raise ToolRegistrationError(f"wildcard permission {n!r} is forbidden (spec §5 T-3)")
        self._allowlists[agent_name] = names

    # -- authorization / invocation -------------------------------------
    def is_authorized(self, agent_name: str, tool_name: str) -> bool:
        return tool_name in self._allowlists.get(agent_name, frozenset())

    def allowlist(self, agent_name: str) -> frozenset[str]:
        return self._allowlists.get(agent_name, frozenset())

    def side_effect(self, tool_name: str) -> str | None:
        """The declared side-effect class of a tool (spec §5 T-5), or None if unregistered."""
        tool = self._tools.get(tool_name)
        return tool.side_effect if tool is not None else None

    def invoke(self, agent_name: str, tool_name: str, arguments: dict[str, Any]) -> Any:
        """Authorize (default-deny) then invoke. Unauthorized → ToolNotAuthorizedError (spec §5 T-1/T-2)."""
        if not self.is_authorized(agent_name, tool_name):
            raise ToolNotAuthorizedError(agent_name, tool_name)
        tool = self._tools.get(tool_name)
        if tool is None:
            raise ToolNotAuthorizedError(agent_name, tool_name)
        return tool.impl(arguments)

grant 

grant(agent_name, tool_names)

Set an agent's explicit tool allowlist (spec §5 T-1, T-3). Wildcards are rejected.

Source code in src/halo_agent_harness/core/registry.py 
57
58
59
60
61
62
63
def grant(self, agent_name: str, tool_names: frozenset[str] | set[str] | list[str]) -> None:
    """Set an agent's explicit tool allowlist (spec §5 T-1, T-3). Wildcards are rejected."""
    names = frozenset(tool_names)
    for n in names:
        if "*" in n:
            raise ToolRegistrationError(f"wildcard permission {n!r} is forbidden (spec §5 T-3)")
    self._allowlists[agent_name] = names

side_effect 

side_effect(tool_name)

The declared side-effect class of a tool (spec §5 T-5), or None if unregistered.

Source code in src/halo_agent_harness/core/registry.py 
72
73
74
75
def side_effect(self, tool_name: str) -> str | None:
    """The declared side-effect class of a tool (spec §5 T-5), or None if unregistered."""
    tool = self._tools.get(tool_name)
    return tool.side_effect if tool is not None else None

invoke 

invoke(agent_name, tool_name, arguments)

Authorize (default-deny) then invoke. Unauthorized → ToolNotAuthorizedError (spec §5 T-1/T-2).

Source code in src/halo_agent_harness/core/registry.py 
77
78
79
80
81
82
83
84
def invoke(self, agent_name: str, tool_name: str, arguments: dict[str, Any]) -> Any:
    """Authorize (default-deny) then invoke. Unauthorized → ToolNotAuthorizedError (spec §5 T-1/T-2)."""
    if not self.is_authorized(agent_name, tool_name):
        raise ToolNotAuthorizedError(agent_name, tool_name)
    tool = self._tools.get(tool_name)
    if tool is None:
        raise ToolNotAuthorizedError(agent_name, tool_name)
    return tool.impl(arguments)

halo_agent_harness.core.side_effect

Side-effect gating policy (spec §5 T-5, §5.3, ADR-0011).

Every registered tool declares a side-effect class: none | read | write | external. Before a tool executes, the harness consults this policy. none/read calls are ungated (they observe, they do not act). write/external calls are gated: they MUST clear a confidence threshold and MUST NOT be issued by a read-only (OBSERVE-authority) agent. A gated call that fails the policy is refused before execution and recorded as a security event — the side effect never happens.

This is the tool-call analogue of the confidence gate (§4): the gate decides auto_enforced for a decision; this policy decides whether a side-effecting tool call may run. It is deliberately small; the general PolicyPort (spec §7) is a later increment.

SideEffectPolicy dataclass

Confidence thresholds for gated side-effect classes. Never below the gate's base floor.

Source code in src/halo_agent_harness/core/side_effect.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass(frozen=True)
class SideEffectPolicy:
    """Confidence thresholds for gated side-effect classes. Never below the gate's base floor."""

    write_threshold: float = 0.85
    external_threshold: float = 0.95

    @staticmethod
    def is_gated(side_effect: str | None) -> bool:
        return side_effect in GATED_CLASSES

    def threshold_for(self, side_effect: str) -> float:
        if side_effect == "write":
            return self.write_threshold
        if side_effect == "external":
            return self.external_threshold
        return 0.0

    def permits(
        self,
        side_effect: str | None,
        confidence: float | None,
        authority: AuthorityLevel,
    ) -> bool:
        """True if a call with this side-effect class may execute (spec §5.3).

        Ungated classes always pass. For gated classes: a read-only (``OBSERVE``) agent is never
        permitted, and the supplied ``confidence`` (absent → 0.0) MUST meet the class threshold.
        """
        if not self.is_gated(side_effect):
            return True
        # OBSERVE agents are read-only — they never perform a write/external side effect.
        if authority <= AuthorityLevel.OBSERVE:
            return False
        c = confidence if confidence is not None else 0.0
        return c >= self.threshold_for(side_effect)  # type: ignore[arg-type]

permits 

permits(side_effect, confidence, authority)

True if a call with this side-effect class may execute (spec §5.3).

Ungated classes always pass. For gated classes: a read-only (OBSERVE) agent is never permitted, and the supplied confidence (absent → 0.0) MUST meet the class threshold.

Source code in src/halo_agent_harness/core/side_effect.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def permits(
    self,
    side_effect: str | None,
    confidence: float | None,
    authority: AuthorityLevel,
) -> bool:
    """True if a call with this side-effect class may execute (spec §5.3).

    Ungated classes always pass. For gated classes: a read-only (``OBSERVE``) agent is never
    permitted, and the supplied ``confidence`` (absent → 0.0) MUST meet the class threshold.
    """
    if not self.is_gated(side_effect):
        return True
    # OBSERVE agents are read-only — they never perform a write/external side effect.
    if authority <= AuthorityLevel.OBSERVE:
        return False
    c = confidence if confidence is not None else 0.0
    return c >= self.threshold_for(side_effect)  # type: ignore[arg-type]

halo_agent_harness.core.harness

The Harness — safe, governed invocation of a single agent (spec §2–§8).

Every agent invocation flows through here. The harness validates scope, honours the kill switch, runs the agent with a registry-enforced tool invoker, applies the confidence gate, routes non-enforcing decisions to human review, and records audit + observability. On any failure it resolves to a safe, non-enforcing decision (spec §8) — it never fails open.

Harness

Wraps agent execution with the harness's controls.

Source code in src/halo_agent_harness/core/harness.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class Harness:
    """Wraps agent execution with the harness's controls."""

    def __init__(
        self,
        registry: ToolRegistry | None = None,
        *,
        audit: AuditPort | None = None,
        human_review: HumanReviewPort | None = None,
        observability: ObservabilityPort | None = None,
        kill_switch: KillSwitchPort | None = None,
        gate: ConfidenceGate | None = None,
        side_effect_policy: SideEffectPolicy | None = None,
    ) -> None:
        # Lazily import the in-memory reference adapters so the core stays import-pure (INV-5):
        # nothing here is imported at module load; only when a port is left unset.
        if audit is None or human_review is None or observability is None or kill_switch is None:
            from ..adapters.inmemory import (
                InMemoryAudit,
                InMemoryHumanReview,
                InMemoryKillSwitch,
                InMemoryObservability,
            )

            audit = audit or InMemoryAudit()
            human_review = human_review or InMemoryHumanReview()
            observability = observability or InMemoryObservability()
            kill_switch = kill_switch or InMemoryKillSwitch()

        self.registry = registry or ToolRegistry()
        self.audit = audit
        self.human_review = human_review
        self.observability = observability
        self.kill_switch = kill_switch
        self.gate = gate or ConfidenceGate()
        self.side_effect_policy = side_effect_policy or SideEffectPolicy()

    # -- public API ------------------------------------------------------
    def invoke(self, agent: Agent, request: AgentInput) -> AgentOutput:
        # Scope is validated, never defaulted (spec §2.1).
        if not request.is_scoped():
            raise UnscopedInvocationError("AgentInput MUST carry non-empty tenantId and userId")

        started = time.perf_counter()

        # Kill switch: short-circuit before any tool side effect; route to human review (spec §7.6).
        if self.kill_switch.is_engaged():
            decision = Decision(
                action=DecisionAction.DEFER,
                confidence=0.0,
                rationale="kill switch engaged — routed to human review",
                auto_enforced=False,
            )
            return self._finalize(agent, request, decision, started, reason="kill_switch")

        decision, reason = self._run_agent(agent, request)
        return self._finalize(agent, request, decision, started, reason=reason)

    # -- internals -------------------------------------------------------
    def _run_agent(self, agent: Agent, request: AgentInput) -> tuple[Decision, str | None]:
        invoker = _BoundInvoker(self, agent.name, agent.authority_level, request)
        try:
            decision = agent.run(request, invoker)
        except ToolNotAuthorizedError:
            return default_for(FailureMode.TOOL_FAILURE, "unauthorized tool call"), "failure"
        except SideEffectDeniedError as exc:
            return default_for(FailureMode.TOOL_FAILURE, f"side-effect denied: {exc.tool_name}"), "failure"
        except Exception as exc:  # any agent/LLM failure resolves safely (spec §8)
            return default_for(FailureMode.BAD_OUTPUT, type(exc).__name__), "failure"

        # The agent must not set auto_enforced; the harness owns it (spec §2.2). Reset defensively.
        decision.auto_enforced = False

        # Undeclared action → contract violation → safe default (spec §10 capability declaration).
        if decision.action not in agent.capabilities:
            return default_for(FailureMode.BAD_OUTPUT, f"undeclared action {decision.action.value}"), "failure"

        # Out-of-authority action → safe default + security-relevant (spec §3.3).
        if not action_within_authority(decision.action, agent.authority_level):
            self.audit.record_security_event(
                SecurityEvent(
                    agent_name=agent.name,
                    tenant_id=request.tenant_id,
                    kind="authority_violation",
                    detail=f"action={decision.action.value} authority={agent.authority_level.name}",
                    correlation_id=request.correlation_id,
                    recorded_at=_now(),
                )
            )
            return default_for(FailureMode.OUT_OF_AUTHORITY, decision.action.value), "failure"

        # Shape validation (confidence range, rationale on BLOCK/ALERT).
        try:
            decision.validate()
        except ValueError as exc:
            return default_for(FailureMode.BAD_OUTPUT, str(exc)), "failure"

        return decision, None

    def _finalize(
        self,
        agent: Agent,
        request: AgentInput,
        decision: Decision,
        started: float,
        *,
        reason: str | None,
    ) -> AgentOutput:
        # The gate is the ONLY place auto_enforced is decided (spec §4, INV-1).
        self.gate.evaluate(decision, agent.authority_level)

        # Defensive bypass detection — must never fire in a correct system (spec §4.2).
        if self.gate.is_bypass(decision, agent.authority_level):
            self.observability.increment_counter(BYPASS_COUNTER)
            decision.auto_enforced = False  # never let a bypass stand

        outcome = "auto-enforced" if decision.auto_enforced else "human-review"

        if not decision.auto_enforced:
            self.human_review.enqueue(
                ReviewItem(
                    agent_name=agent.name,
                    request=request,
                    decision=decision,
                    reason=reason or self._review_reason(decision),
                    sla_seconds=_SLA_BLOCK if decision.action is DecisionAction.BLOCK else _SLA_DEFAULT,
                    enqueued_at=_now(),
                )
            )

        self.audit.record(
            AuditEntry(
                agent_name=agent.name,
                tenant_id=request.tenant_id,
                action=decision.action.value,
                confidence=decision.confidence,
                auto_enforced=decision.auto_enforced,
                rationale=decision.rationale,
                outcome=outcome,
                correlation_id=request.correlation_id,
                recorded_at=_now(),
            )
        )

        duration_ms = (time.perf_counter() - started) * 1000.0
        self.observability.emit(
            InvocationMetric(
                agent_name=agent.name,
                action=decision.action.value,
                confidence=decision.confidence,
                duration_ms=duration_ms,
                outcome=outcome,
                correlation_id=request.correlation_id,
            )
        )
        return AgentOutput.now(decision, agent.name)

    @staticmethod
    def _review_reason(decision: Decision) -> str:
        if decision.action is DecisionAction.DEFER:
            return "defer"
        if decision.action is DecisionAction.SUGGEST:
            return "suggest"
        return "low_confidence"

halo_agent_harness.core.failure

Failure-mode defaults (spec §8).

Every failure resolves to a safe decision: lowered confidence and auto_enforced=False. An agent never fails open into an autonomous action.

default_for 

default_for(mode, detail='')

Return the safe fallback decision for a failure mode (spec §8).

Source code in src/halo_agent_harness/core/failure.py 
34
35
36
37
38
39
40
41
def default_for(mode: FailureMode, detail: str = "") -> Decision:
    """Return the safe fallback decision for a failure mode (spec §8)."""
    action, confidence = _DEFAULTS[mode]
    rationale = f"harness failure default ({mode.value})"
    if detail:
        rationale = f"{rationale}: {detail}"
    # auto_enforced is False and stays False — the gate will confirm it.
    return Decision(action=action, confidence=confidence, rationale=rationale, auto_enforced=False)

Orchestration (spec §6)

halo_agent_harness.orchestration.pipeline

Sequential Pipeline orchestration (spec §6.1).

Agents run in order. Each stage receives the prior stage's decision in its context under the pipeline key, so a later stage can react to what came before. Every stage is invoked through Harness.invoke, so each passes the confidence gate and tool registry individually (O-1).

The pipeline short-circuits on the first BLOCK or DEFER (spec §6.1): once a stage refuses or defers, later stages do not run — the safe action stands and no further side effects are risked.

PipelineResult dataclass

The outcome of a sequential pipeline run.

Source code in src/halo_agent_harness/orchestration/pipeline.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class PipelineResult:
    """The outcome of a sequential pipeline run."""

    final_action: DecisionAction
    stage_outputs: dict[str, AgentOutput]
    short_circuited_at: str | None = None

    @property
    def reconciled_action(self) -> DecisionAction:
        """The safest action across every stage that actually ran (spec §3.3)."""
        return reconcile([o.decision.action for o in self.stage_outputs.values()])

    @property
    def any_auto_enforced(self) -> bool:
        return any(o.decision.auto_enforced for o in self.stage_outputs.values())

reconciled_action property 

reconciled_action

The safest action across every stage that actually ran (spec §3.3).

Pipeline

Runs agents sequentially, feeding each stage the prior stage's decision.

Source code in src/halo_agent_harness/orchestration/pipeline.py 
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
class Pipeline:
    """Runs agents sequentially, feeding each stage the prior stage's decision."""

    def __init__(self, harness: Harness, stages: list[Agent]) -> None:
        if not stages:
            raise ValueError("a pipeline requires at least one stage")
        self._harness = harness
        self._stages = list(stages)

    def run(self, request: AgentInput) -> PipelineResult:
        outputs: dict[str, AgentOutput] = {}
        short_circuited_at: str | None = None
        current = request

        for stage in self._stages:
            output = self._harness.invoke(stage, current)
            outputs[stage.name] = output
            action = output.decision.action
            if action in _SHORT_CIRCUIT:
                short_circuited_at = stage.name
                break
            current = _with_prior(request, stage.name, output)

        final_action = outputs[list(outputs)[-1]].decision.action
        return PipelineResult(
            final_action=final_action,
            stage_outputs=outputs,
            short_circuited_at=short_circuited_at,
        )

halo_agent_harness.orchestration.fanout

Parallel Fan-out orchestration (spec §6.2).

Independent worker agents run concurrently over the same AgentInput; the harness reconciles their decisions with the Decision Hierarchy (BLOCK wins, then ALERT, …). Every worker is invoked through Harness.invoke, so each passes the confidence gate and tool registry individually (O-1). The aggregation policy is collect-and-reconcile — every worker runs and the safest action wins; no worker's controls are skipped.

Concurrency: workers run on a thread pool. The reference in-memory adapters are concurrency-safe (the bypass counter is incremented under a lock), so confidence_gate_bypass_total accumulates correctly.

FanOutResult dataclass

The aggregated outcome of a parallel fan-out.

Source code in src/halo_agent_harness/orchestration/fanout.py 
23
24
25
26
27
28
29
30
31
32
@dataclass(frozen=True)
class FanOutResult:
    """The aggregated outcome of a parallel fan-out."""

    reconciled_action: DecisionAction
    worker_outputs: dict[str, AgentOutput]

    @property
    def any_auto_enforced(self) -> bool:
        return any(o.decision.auto_enforced for o in self.worker_outputs.values())

FanOut

Runs independent workers concurrently over one input, then reconciles their decisions.

Source code in src/halo_agent_harness/orchestration/fanout.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class FanOut:
    """Runs independent workers concurrently over one input, then reconciles their decisions."""

    def __init__(self, harness: Harness, workers: list[Agent], *, max_workers: int | None = None) -> None:
        if not workers:
            raise ValueError("fan-out requires at least one worker agent")
        self._harness = harness
        self._workers = list(workers)
        self._max_workers = max_workers or len(workers)

    def run(self, request: AgentInput) -> FanOutResult:
        # Submit every worker; collect results keyed by worker so the mapping is order-stable
        # regardless of completion order.
        with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
            futures = {w.name: pool.submit(self._harness.invoke, w, request) for w in self._workers}
            outputs = {name: future.result() for name, future in futures.items()}

        reconciled = reconcile([o.decision.action for o in outputs.values()])
        return FanOutResult(reconciled_action=reconciled, worker_outputs=outputs)

halo_agent_harness.orchestration.debate

Debate / Consensus orchestration (spec §6.4).

Multiple agents produce competing decisions over the same AgentInput; a consensus rule reconciles them. Every participant is invoked through Harness.invoke, so each passes the confidence gate and tool registry individually (O-1).

Two consensus rules are provided:

  • SAFEST (default) — the strictest action wins, per the Decision Hierarchy (§3.3). Conservative: a single BLOCK beats any number of ALLOWs. This is the safe default for governance.
  • MAJORITY — the most-proposed action wins; a tie resolves to DEFER (tie → human review). MAJORITY may de-escalate below the strictest proposal (that is the point of a vote); use SAFEST when the strictest participant must always win.

Safety floor (invariant). Consensus can never raise authority above the strictest participant: the consensus action's severity never exceeds the strictest action any participant actually proposed. Since every rule chooses among proposed actions, this holds by construction; it is enforced defensively and asserted by tests.

DebateResult dataclass

The reconciled outcome of a debate.

Source code in src/halo_agent_harness/orchestration/debate.py 
43
44
45
46
47
48
49
50
51
52
53
54
@dataclass(frozen=True)
class DebateResult:
    """The reconciled outcome of a debate."""

    consensus_action: DecisionAction
    rule: ConsensusRule
    participant_outputs: dict[str, AgentOutput]
    tie: bool = False

    @property
    def any_auto_enforced(self) -> bool:
        return any(o.decision.auto_enforced for o in self.participant_outputs.values())

Debate

Runs competing agents and reconciles their decisions with a consensus rule.

Source code in src/halo_agent_harness/orchestration/debate.py 
 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
class Debate:
    """Runs competing agents and reconciles their decisions with a consensus rule."""

    def __init__(
        self,
        harness: Harness,
        participants: list[Agent],
        *,
        rule: ConsensusRule = ConsensusRule.SAFEST,
    ) -> None:
        if not participants:
            raise ValueError("a debate requires at least one participant")
        self._harness = harness
        self._participants = list(participants)
        self._rule = rule

    def run(self, request: AgentInput) -> DebateResult:
        outputs = {p.name: self._harness.invoke(p, request) for p in self._participants}
        actions = [o.decision.action for o in outputs.values()]

        action, tie = self._consensus(actions)

        # Safety floor (§6.4): never exceed the strictest action any participant proposed.
        ceiling = reconcile(actions)
        if action_precedence(action) > action_precedence(ceiling):
            action = ceiling

        return DebateResult(
            consensus_action=action,
            rule=self._rule,
            participant_outputs=outputs,
            tie=tie,
        )

    def _consensus(self, actions: list[DecisionAction]) -> tuple[DecisionAction, bool]:
        if self._rule is ConsensusRule.SAFEST:
            return reconcile(actions), False

        # MAJORITY: highest vote count wins; a tie for the top count resolves to DEFER.
        counts = Counter(actions)
        top = max(counts.values())
        winners = [a for a, c in counts.items() if c == top]
        if len(winners) == 1:
            return winners[0], False
        return DecisionAction.DEFER, True

halo_agent_harness.orchestration.supervisor

Supervisor + Workers orchestration (spec §6.3, ADR-0007) — the primary multi-step pattern.

A supervisor coordinates worker agents. The supervisor holds NO tool permissions (T-4). The run has two phases:

  1. Planning turn. The supervisor is invoked through the harness (Harness.invoke), so its planning decision passes the confidence gate, kill switch, and audit like any other invocation (O-1) — it just cannot touch tools. If the supervisor's decision is BLOCK or DEFER it halts delegation: no workers run. Otherwise the supervisor may select which workers to engage.
  2. Delegation. Selected workers are each invoked through the harness (gate + registry individually), and their decisions are reconciled via the Decision Hierarchy (spec §3.3).

Worker selection is optional: a supervisor that also implements :class:Planner chooses a subset of workers; a plain supervisor delegates to all of them (backwards-compatible).

Planner

Bases: Protocol

Optional supervisor capability: choose which workers to delegate to (spec §6.3).

Planning is a pure coordination step — it selects names, it does not act — so it needs no tools.

Source code in src/halo_agent_harness/orchestration/supervisor.py 
31
32
33
34
35
36
37
38
39
40
@runtime_checkable
class Planner(Protocol):
    """Optional supervisor capability: choose which workers to delegate to (spec §6.3).

    Planning is a pure coordination step — it selects names, it does not act — so it needs no tools.
    """

    def plan(self, request: AgentInput, worker_names: list[str]) -> list[str]:
        """Return the subset of ``worker_names`` to engage for this request."""
        ...

plan 

plan(request, worker_names)

Return the subset of worker_names to engage for this request.

Source code in src/halo_agent_harness/orchestration/supervisor.py 
38
39
40
def plan(self, request: AgentInput, worker_names: list[str]) -> list[str]:
    """Return the subset of ``worker_names`` to engage for this request."""
    ...

SupervisorWorkers

Coordinates worker agents under a tool-less supervisor with a real planning turn.

Source code in src/halo_agent_harness/orchestration/supervisor.py 
 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
class SupervisorWorkers:
    """Coordinates worker agents under a tool-less supervisor with a real planning turn."""

    def __init__(self, harness: Harness, supervisor: Agent, workers: list[Agent]) -> None:
        if harness.registry.allowlist(supervisor.name):
            raise ToolRegistrationError(
                f"supervisor {supervisor.name!r} MUST hold no tool permissions (spec §6.3 T-4)"
            )
        if not workers:
            raise ValueError("at least one worker agent is required")
        self._harness = harness
        self._supervisor = supervisor
        self._workers = workers

    def run(self, request: AgentInput) -> OrchestrationResult:
        # Phase 1 — governed planning turn. The supervisor reasons about the task under the harness.
        supervisor_output = self._harness.invoke(self._supervisor, request)
        plan_action = supervisor_output.decision.action

        # A halting supervisor decision stops the orchestration before any worker side effect.
        if plan_action in _HALT:
            return OrchestrationResult(
                reconciled_action=plan_action,
                worker_outputs={},
                supervisor_output=supervisor_output,
                delegated=(),
                halted=True,
            )

        # Phase 2 — delegation. Select workers (all, unless the supervisor is a Planner).
        workers_by_name = {w.name: w for w in self._workers}
        selected = self._select(request, list(workers_by_name))

        outputs: dict[str, AgentOutput] = {}
        for name in selected:
            outputs[name] = self._harness.invoke(workers_by_name[name], request)

        # Reconcile the workers that acted; with no delegation, the supervisor's own action stands.
        actions = [o.decision.action for o in outputs.values()] or [plan_action]
        reconciled = reconcile(actions)
        return OrchestrationResult(
            reconciled_action=reconciled,
            worker_outputs=outputs,
            supervisor_output=supervisor_output,
            delegated=tuple(selected),
            halted=False,
        )

    def _select(self, request: AgentInput, worker_names: list[str]) -> list[str]:
        if isinstance(self._supervisor, Planner):
            chosen = self._supervisor.plan(request, list(worker_names))
            # Constrain to real workers; the supervisor cannot invent or reorder beyond the roster.
            return [n for n in chosen if n in worker_names]
        return worker_names

Ports (spec §7)

halo_agent_harness.ports.governance

Governance ports — audit, human review, observability, memory, kill switch, policy (spec §7).

All are Protocols; concrete adapters live at the edge and depend on this core, never the reverse (ADR-0005). Data types carried across the ports are defined here.

AuditPort

Bases: Protocol

Append-only audit log. Implementations MUST NOT update or delete, and MUST redact PII.

Source code in src/halo_agent_harness/ports/governance.py 
42
43
44
45
46
47
@runtime_checkable
class AuditPort(Protocol):
    """Append-only audit log. Implementations MUST NOT update or delete, and MUST redact PII."""

    def record(self, entry: AuditEntry) -> None: ...
    def record_security_event(self, event: SecurityEvent) -> None: ...

ReviewItem dataclass

Source code in src/halo_agent_harness/ports/governance.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@dataclass(frozen=True)
class ReviewItem:
    agent_name: str
    request: AgentInput
    decision: Decision
    reason: str             # why it was routed (low_confidence | defer | suggest | kill_switch | failure)
    sla_seconds: int
    enqueued_at: datetime

    @property
    def deadline(self) -> datetime:
        """When this item breaches its SLA (spec §7.4)."""
        return self.enqueued_at + timedelta(seconds=self.sla_seconds)

    def is_overdue(self, now: datetime) -> bool:
        return now > self.deadline

deadline property 

deadline

When this item breaches its SLA (spec §7.4).

OverrideRecord dataclass

A human override of a queued decision — itself auditable (spec §7.4).

Source code in src/halo_agent_harness/ports/governance.py 
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class OverrideRecord:
    """A human override of a queued decision — itself auditable (spec §7.4)."""

    review_id: int
    agent_name: str
    tenant_id: str
    reviewer: str
    outcome: str            # the human's resolution, e.g. "approved" | "rejected" | "amended"
    resolved_at: datetime

HumanReviewPort

Bases: Protocol

Queue for decisions that must reach a human, with an SLA per item.

Source code in src/halo_agent_harness/ports/governance.py 
83
84
85
86
87
@runtime_checkable
class HumanReviewPort(Protocol):
    """Queue for decisions that must reach a human, with an SLA per item."""

    def enqueue(self, item: ReviewItem) -> None: ...

ObservabilityPort

Bases: Protocol

Per-invocation metrics + counters. MUST expose confidence_gate_bypass_total (must stay 0).

Source code in src/halo_agent_harness/ports/governance.py 
103
104
105
106
107
108
@runtime_checkable
class ObservabilityPort(Protocol):
    """Per-invocation metrics + counters. MUST expose ``confidence_gate_bypass_total`` (must stay 0)."""

    def emit(self, metric: InvocationMetric) -> None: ...
    def increment_counter(self, name: str, value: int = 1) -> None: ...

KillSwitchPort

Bases: Protocol

System-wide stop without a code deploy. When engaged, nothing auto-enforces.

Source code in src/halo_agent_harness/ports/governance.py 
114
115
116
117
118
@runtime_checkable
class KillSwitchPort(Protocol):
    """System-wide stop without a code deploy. When engaged, nothing auto-enforces."""

    def is_engaged(self) -> bool: ...

halo_agent_harness.ports.llm

LlmPort — provider-agnostic LLM access (spec §7).

The shape is adopted verbatim from the apex-sdlc provider abstraction (apex-sdlc/platform/backend/app/integrations/llm/base.py) so existing adapters plug straight in. No agent calls an SDK directly; all LLM access goes through this port (INV: Ollama/provider replaceable).

Message dataclass

A single chat message.

Source code in src/halo_agent_harness/ports/llm.py 
14
15
16
17
18
19
@dataclass
class Message:
    """A single chat message."""

    role: str  # "system" | "user" | "assistant"
    content: str

ToolDefinition dataclass

Describes a callable tool that can be offered to an LLM (spec §5.2).

Source code in src/halo_agent_harness/ports/llm.py 
22
23
24
25
26
27
28
29
@dataclass
class ToolDefinition:
    """Describes a callable tool that can be offered to an LLM (spec §5.2)."""

    name: str
    description: str
    parameters: dict  # JSON Schema object
    side_effect: str = "read"  # none | read | write | external

ToolCall dataclass

A tool invocation requested by the LLM.

Source code in src/halo_agent_harness/ports/llm.py 
32
33
34
35
36
37
38
@dataclass
class ToolCall:
    """A tool invocation requested by the LLM."""

    name: str
    arguments: dict
    id: str = ""

CompletionResult dataclass

Unified response from any LLM provider.

Source code in src/halo_agent_harness/ports/llm.py 
41
42
43
44
45
46
47
48
49
50
@dataclass
class CompletionResult:
    """Unified response from any LLM provider."""

    content: str
    tool_calls: list[ToolCall] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    model: str = ""
    provider: str = ""

LlmPort

Bases: Protocol

Protocol every LLM provider adapter must satisfy.

Source code in src/halo_agent_harness/ports/llm.py 
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
@runtime_checkable
class LlmPort(Protocol):
    """Protocol every LLM provider adapter must satisfy."""

    provider_name: str
    model: str

    async def complete(
        self,
        messages: list[Message],
        system: str | None = None,
        tools: list[ToolDefinition] | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.7,
    ) -> CompletionResult:
        """Generate a single completion and return the full result."""
        ...

    async def stream(
        self,
        messages: list[Message],
        system: str | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.7,
    ) -> AsyncIterator[str]:
        """Stream text tokens as they are generated."""
        ...

complete async 

complete(messages, system=None, tools=None, max_tokens=4096, temperature=0.7)

Generate a single completion and return the full result.

Source code in src/halo_agent_harness/ports/llm.py 
60
61
62
63
64
65
66
67
68
69
async def complete(
    self,
    messages: list[Message],
    system: str | None = None,
    tools: list[ToolDefinition] | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
) -> CompletionResult:
    """Generate a single completion and return the full result."""
    ...

stream async 

stream(messages, system=None, max_tokens=4096, temperature=0.7)

Stream text tokens as they are generated.

Source code in src/halo_agent_harness/ports/llm.py 
71
72
73
74
75
76
77
78
79
async def stream(
    self,
    messages: list[Message],
    system: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
) -> AsyncIterator[str]:
    """Stream text tokens as they are generated."""
    ...

Contract loading (spec §10)

halo_agent_harness.contract

Agent Contract loading + validation (spec §10, ADR-0006).

Validates a contract against docs/spec/agent-contract.schema.json and then applies the semantic binding rule (declared capabilities must be within the declared authority level, spec §3.3) that the JSON Schema alone cannot express. Requires the contract extra (jsonschema).

validate_contract 

validate_contract(contract, schema_path=None)

Validate a contract dict; return it unchanged on success, else raise ContractValidationError.

Source code in src/halo_agent_harness/contract.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def validate_contract(contract: dict[str, Any], schema_path: str | Path | None = None) -> dict[str, Any]:
    """Validate a contract dict; return it unchanged on success, else raise ContractValidationError."""
    try:
        import jsonschema
    except ImportError as exc:  # pragma: no cover - environment guard
        raise ContractValidationError(
            "jsonschema is required to validate contracts (install the 'contract' extra)"
        ) from exc

    schema = load_schema(schema_path)
    validator = jsonschema.Draft7Validator(schema)
    errors = sorted(validator.iter_errors(contract), key=lambda e: list(e.path))
    if errors:
        first = errors[0]
        raise ContractValidationError(f"{list(first.path)}: {first.message}")

    # Semantic binding rule (spec §3.3): every declared capability must be within authorityLevel.
    authority = AuthorityLevel[contract["authorityLevel"]]
    for cap in contract["capabilities"]:
        if not action_within_authority(DecisionAction[cap], authority):
            raise ContractValidationError(
                f"capability {cap} exceeds declared authority {authority.name} (spec §3.3)"
            )
    return contract

load_contract 

load_contract(path, schema_path=None)

Load a contract JSON file and validate it (spec §10).

Source code in src/halo_agent_harness/contract.py 
59
60
61
62
def load_contract(path: str | Path, schema_path: str | Path | None = None) -> dict[str, Any]:
    """Load a contract JSON file and validate it (spec §10)."""
    contract = json.loads(Path(path).read_text())
    return validate_contract(contract, schema_path)

Reference adapters

halo_agent_harness.adapters.sla

Human-review SLA monitor (spec §7.4).

Sweeps a review queue for items past their SLA deadline and emits a breach counter through the ObservabilityPort. Each breached item is counted at most once, so repeated sweeps are idempotent. This is the enforcement/monitoring hook the spec calls for; escalation policy (paging, reassignment) is the integrator's concern and sits above this signal.

SlaMonitor

Counts SLA breaches in a human-review queue, idempotently across sweeps.

Source code in src/halo_agent_harness/adapters/sla.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class SlaMonitor:
    """Counts SLA breaches in a human-review queue, idempotently across sweeps."""

    def __init__(self, queue: _OverdueSource, observability: ObservabilityPort) -> None:
        self._queue = queue
        self._obs = observability
        self._counted: set[int] = set()

    def sweep(self, now: datetime | None = None) -> tuple:
        """Return currently-overdue items; increment the breach counter for newly-breached ones."""
        breached = self._queue.overdue(now)
        for q in breached:
            if q.id not in self._counted:
                self._counted.add(q.id)
                self._obs.increment_counter(HUMAN_REVIEW_SLA_BREACH)
        return breached

    @property
    def breach_count(self) -> int:
        return len(self._counted)

sweep 

sweep(now=None)

Return currently-overdue items; increment the breach counter for newly-breached ones.

Source code in src/halo_agent_harness/adapters/sla.py 
32
33
34
35
36
37
38
39
def sweep(self, now: datetime | None = None) -> tuple:
    """Return currently-overdue items; increment the breach counter for newly-breached ones."""
    breached = self._queue.overdue(now)
    for q in breached:
        if q.id not in self._counted:
            self._counted.add(q.id)
            self._obs.increment_counter(HUMAN_REVIEW_SLA_BREACH)
    return breached

halo_agent_harness.adapters.redaction

Pluggable PII/secret redaction (spec §7.3).

Redaction runs before every audit write; zero PII in logs is a P1 condition. The default strategy covers common patterns (JWT, email, card, SSN, phone, API key) — but every deployment has its own sensitive shapes, so the strategy is pluggable: extend it with with_rule or build one from scratch. A RedactionStrategy is callable, so it drops directly into any port that takes a redactor.

The default patterns are best-effort, not a guarantee: order matters (specific before generic) and no regex set is exhaustive. Treat redaction as defense-in-depth, not the only control.

RedactionStrategy dataclass

An ordered, immutable set of redaction rules. Callable so it plugs in wherever a redactor is asked.

Source code in src/halo_agent_harness/adapters/redaction.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@dataclass(frozen=True)
class RedactionStrategy:
    """An ordered, immutable set of redaction rules. Callable so it plugs in wherever a redactor is asked."""

    rules: tuple[RedactionRule, ...] = field(default=_DEFAULT_RULES)

    def redact(self, text: str) -> str:
        out = text
        for rule in self.rules:
            out = rule.pattern.sub(rule.replacement, out)
        return out

    def __call__(self, text: str) -> str:
        return self.redact(text)

    def with_rule(self, pattern: str | re.Pattern[str], replacement: str) -> "RedactionStrategy":
        """Return a new strategy with one extra rule appended (the base stays immutable)."""
        compiled = re.compile(pattern) if isinstance(pattern, str) else pattern
        return replace(self, rules=self.rules + (RedactionRule(compiled, replacement),))

with_rule 

with_rule(pattern, replacement)

Return a new strategy with one extra rule appended (the base stays immutable).

Source code in src/halo_agent_harness/adapters/redaction.py 
51
52
53
54
def with_rule(self, pattern: str | re.Pattern[str], replacement: str) -> "RedactionStrategy":
    """Return a new strategy with one extra rule appended (the base stays immutable)."""
    compiled = re.compile(pattern) if isinstance(pattern, str) else pattern
    return replace(self, rules=self.rules + (RedactionRule(compiled, replacement),))

redact 

redact(text)

Redact with the default strategy (spec §7.3). Kept as a free function for convenience.

Source code in src/halo_agent_harness/adapters/redaction.py 
60
61
62
def redact(text: str) -> str:
    """Redact with the default strategy (spec §7.3). Kept as a free function for convenience."""
    return DEFAULT_STRATEGY.redact(text)

halo_agent_harness.adapters.policy

PolicyPort reference adapter (spec §7) — evaluate action against immutable rules.

Rules are fixed at construction (frozen dataclasses): a policy cannot be mutated at runtime, which upholds the no-self-escalation invariant (INV-3) — an agent cannot loosen the rules that bind it. The first matching rule wins; if none match, the configured default applies. load_policy builds the same immutable policy from a durable JSON file.

PolicyRule dataclass

A single allow/deny rule. None in a field means 'any' (not a wildcard string — an explicit any).

Source code in src/halo_agent_harness/adapters/policy.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass(frozen=True)
class PolicyRule:
    """A single allow/deny rule. ``None`` in a field means 'any' (not a wildcard string — an explicit any)."""

    effect: str                     # "allow" | "deny"
    agent: str | None = None
    action: str | None = None
    tenant: str | None = None

    def __post_init__(self) -> None:
        if self.effect not in ("allow", "deny"):
            raise ValueError(f"rule effect must be 'allow' or 'deny', got {self.effect!r}")

    def matches(self, agent_name: str, action: str, tenant_id: str) -> bool:
        return (
            (self.agent is None or self.agent == agent_name)
            and (self.action is None or self.action == action)
            and (self.tenant is None or self.tenant == tenant_id)
        )

RuleBasedPolicy dataclass

Immutable rule set. First matching rule wins; otherwise default_allow decides.

Source code in src/halo_agent_harness/adapters/policy.py 
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass(frozen=True)
class RuleBasedPolicy:
    """Immutable rule set. First matching rule wins; otherwise ``default_allow`` decides."""

    rules: tuple[PolicyRule, ...] = field(default_factory=tuple)
    default_allow: bool = True

    def permits(self, agent_name: str, action: str, tenant_id: str) -> bool:
        for rule in self.rules:
            if rule.matches(agent_name, action, tenant_id):
                return rule.effect == "allow"
        return self.default_allow

load_policy 

load_policy(path)

Load an immutable policy from a JSON file: {"default_allow": bool, "rules": [ {...}, ... ]}.

Source code in src/halo_agent_harness/adapters/policy.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def load_policy(path: str | Path) -> RuleBasedPolicy:
    """Load an immutable policy from a JSON file: ``{"default_allow": bool, "rules": [ {...}, ... ]}``."""
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    rules = tuple(
        PolicyRule(
            effect=r["effect"],
            agent=r.get("agent"),
            action=r.get("action"),
            tenant=r.get("tenant"),
        )
        for r in data.get("rules", [])
    )
    return RuleBasedPolicy(rules=rules, default_allow=bool(data.get("default_allow", True)))