Trajectory Preflight

Trajectory preflight estimates whether a prompt is likely to trigger a runtime halt before the main generation starts. TrajectorySimulator produces seeded candidate trajectories and PredictivePreHaltSteering turns the aggregate verdict into a calibrated allow, warn, or halt guard decision.

Predictive Steering

PredictivePreHaltSteering uses three evidence gates:

• halt when the empirical halt probability crosses the calibrated risk threshold
• escalate when the upper confidence bound crosses the calibrated threshold
• escalate when there are too few simulations for the configured minimum

The steering payload is tenant-safe. It stores halt probability, confidence interval bounds, backend recommendation, and trajectory IDs for failed draws; it does not serialise prompt text or sampled token text.

from director_ai.core.guard_control import RiskEnvelope
from director_ai.core.trajectory import PredictivePreHaltSteering

steering = PredictivePreHaltSteering(min_simulations=8)
decision = steering.evaluate(
    verdict,
    risk_envelope=RiskEnvelope(
        action_category="inference_steering",
        reversibility="reversible",
        domain="regulated",
        calibrated_threshold=0.5,
        no_go_threshold=0.9,
    ),
    policy_id="policy.prehalt.regulated",
)

if decision.action == "escalate":
    route_to = decision.recommended_backend

Rollback Hooks

TrajectoryRollbackManager registers tenant-safe rollback handles before a high-risk action executes. A low-risk preflight leaves the handle unused, a warn or escalation arms the handle for operator review, and a halt executes the hook once. Repeated execution returns already_executed.

from director_ai.core.trajectory import TrajectoryRollbackManager

rollback = TrajectoryRollbackManager()
handle = rollback.register(
    rollback_id="deploy-rollback-20260604-a",
    action_id="deploy-policy-overlay",
    hook=lambda handle, reason: {"rollback_store": "audit-log"},
    evidence_refs=("change:42",),
    metadata={"owner": "safety"},
)
outcome = rollback.evaluate_preflight(handle.rollback_id, verdict)

Rollback payloads contain rollback/action IDs, tenant ID, evidence references, status, and tenant-safe metadata only. They do not include prompt text, sampled token text, raw action bodies, credentials, or sensor payloads.

Use InferenceServerHook.steer() when the decision must affect pre-sampling logits directly. The hook maps escalate to a finite negative bias for the candidate token and maps halt to the same block action used by the coherence threshold hook.

Full API

director_ai.core.trajectory.simulator.TrajectorySimulator

TrajectorySimulator(*, actor: Actor, scorer: VerdictProducer, n_simulations: int = 8, halt_rate_halt: float = 0.5, halt_rate_warn: float = 0.25, base_seed: int = 17, ci_level: float = 0.95)

Pre-execution Monte-Carlo halt check.

Parameters:

Name	Type	Description	Default
actor	Actor	A :class:Actor that samples token sequences.	required
scorer	VerdictProducer	A :class:VerdictProducer that returns (approved, score).	required
n_simulations	int	Number of independent draws per :meth:preflight call. Defaults to 8 — enough for a useful halt-rate estimate without turning preflight into the slow path.	8
halt_rate_halt	float	Halt-rate threshold above which :meth:preflight returns action="halt".	0.5
halt_rate_warn	float	Halt-rate threshold for action="warn". Between this and halt_rate_halt the caller is expected to escalate (e.g. route to a stronger model).	0.25
base_seed	int	Seed offset; per-trajectory seeds are base_seed + i.	17
ci_level	float	Credible-interval level for the ci_low / ci_high fields; 0.95 is the default.	0.95

preflight

preflight(prompt: str, *, tenant_id: str = '', on_trajectory: Callable[[TrajectoryResult], None] | None = None) -> PreflightVerdict

Run the Monte-Carlo loop and return an aggregate verdict.

on_trajectory is an optional per-draw callback — handy for live observability sinks that want to stream intermediate results without waiting for the aggregate.

director_ai.core.trajectory.simulator.PreflightVerdict dataclass

PreflightVerdict(n_simulations: int, halt_rate: float, mean_coherence: float, std_coherence: float, ci_low: float, ci_high: float, recommended: Action, reason: str, trajectories: tuple[TrajectoryResult, ...] = tuple(), safety_event: SafetyEvent | None = None)

Aggregate of N simulated trajectories.

min_coherence property

min_coherence: float

Return the lowest final coherence across the simulated trajectories.

max_coherence property

max_coherence: float

Return the highest final coherence across the simulated trajectories.

director_ai.core.trajectory.steering.PredictivePreHaltSteering

PredictivePreHaltSteering(*, min_simulations: int = 8, uncertainty_margin: float = 0.0)

Escalate or halt requests using calibrated trajectory risk evidence.

evaluate

evaluate(verdict: PreflightVerdict, *, risk_envelope: RiskEnvelope, policy_id: str) -> PreHaltSteeringDecision

Return a pre-halt steering decision from a trajectory verdict.

director_ai.core.trajectory.steering.PreHaltSteeringDecision dataclass

PreHaltSteeringDecision(action: SteeringAction, reason: str, halt_probability: float, ci_low: float, ci_high: float, recommended_backend: str, evidence_refs: Sequence[str], guard_decision: GuardDecision)

Tenant-safe pre-halt steering decision derived from trajectory evidence.

__post_init__

__post_init__() -> None

Reject an unknown steering action and require a non-empty reason.

to_dict

to_dict() -> dict[str, Any]

Serialise without trajectory token text or prompt payloads.

to_safety_event

to_safety_event(*, hook_id: str, hook_scope: str = 'trajectory', request_id: str = '', tenant_id: str = '', latency_ms: float | None = None) -> SafetyEvent

Convert the steering decision to the shared safety-event schema.

director_ai.core.trajectory.rollback.TrajectoryRollbackManager

TrajectoryRollbackManager()

Register and execute rollback hooks from trajectory decisions.

Hooks are idempotent by rollback id: a successful rollback executes once, then later calls return already_executed. The manager stores tenant-safe identifiers, evidence references, and metadata only; raw prompts, sampled tokens, action bodies, credentials, and sensor payloads should stay in the deployment's own protected system of record.

register

register(*, rollback_id: str, action_id: str, hook: RollbackHook, tenant_id: str = '', evidence_refs: Sequence[str] = (), metadata: Mapping[str, str] | None = None) -> RollbackHandle

Register a rollback hook and return its handle.

evaluate_preflight

evaluate_preflight(rollback_id: str, verdict: PreflightVerdict, *, steering_decision: object | None = None) -> RollbackOutcome

Arm or execute rollback according to trajectory evidence.

execute

execute(rollback_id: str, *, reason: str, evidence_refs: Sequence[str] = ()) -> RollbackOutcome

Execute a registered rollback hook once.

director_ai.core.trajectory.rollback.RollbackOutcome dataclass

RollbackOutcome(rollback_id: str, action_id: str, status: RollbackStatus, reason: str, executed: bool, tenant_id: str = '', evidence_refs: Sequence[str] = tuple(), metadata: Mapping[str, str] = dict(), error_type: str = '')

Result of arming or executing a rollback hook.

__post_init__

__post_init__() -> None

Reject an unknown status code and require rollback_id.

to_dict

to_dict() -> dict[str, object]

Return a tenant-safe JSON representation.