harn_vm/
lib.rs

1#![recursion_limit = "256"]
2#![allow(clippy::result_large_err, clippy::cloned_ref_to_slice_refs)]
3//! # harn-vm
4//!
5//! The Harn compiler, virtual machine, standard library, provider/LLM layer,
6//! orchestration runtime, and host bridge.
7//!
8//! ## Stability
9//!
10//! This crate is consumed both by the in-tree surfaces (`harn-cli`,
11//! `harn-serve`, the LSP and DAP) and by external embedders. The intended
12//! embedding entry points are `Vm`, `Harness`, `compile_source`, and the
13//! `llm`, `orchestration`, `agent_events`, `agent_sessions`, `config`, and
14//! `security` modules. Other public items exist primarily for in-workspace use
15//! and may change between minor releases; anything marked `#[doc(hidden)]` is
16//! an implementation detail with no stability guarantee. The crate follows the
17//! workspace version and is pre-1.0, so the public surface may still evolve.
18
19/// Re-export of the unified clock substrate so downstream crates (CLI,
20/// orchestrator, and cloud runtimes) can depend on a single canonical `Clock`
21/// trait without each adding `harn-clock` as a direct dependency.
22pub use harn_clock as clock;
23
24pub mod a2a;
25pub mod actor_chain;
26pub mod agent_events;
27pub mod agent_sessions;
28pub mod atomic_io;
29pub mod autonomy;
30pub(crate) mod aws_sigv4;
31pub mod bridge;
32mod builtin_id;
33pub mod bytecode_cache;
34pub mod call_budget;
35pub mod channel_guardrails;
36pub mod channels;
37pub mod checkpoint;
38mod chunk;
39mod compiler;
40pub mod composition;
41pub mod config;
42pub mod connectors;
43pub mod corrections;
44pub mod coverage;
45pub(crate) mod durable_rate_limit;
46pub(crate) mod duration_parse;
47pub mod egress;
48pub mod event_log;
49pub mod events;
50pub mod external_agent;
51pub mod flow;
52pub mod harness;
53pub mod harness_auth;
54pub(crate) mod harness_crypto;
55pub mod harness_net;
56pub mod harness_system;
57pub mod harness_tenant;
58mod http;
59pub mod jsonrpc;
60pub(crate) mod limits;
61pub mod llm;
62pub mod llm_config;
63pub mod mcp;
64pub mod mcp_allowlist;
65pub mod mcp_auth;
66pub mod mcp_bulk_auth;
67pub mod mcp_card;
68pub mod mcp_elicit;
69pub mod mcp_file_upload;
70pub mod mcp_host;
71pub mod mcp_identity;
72pub mod mcp_json_discovery;
73pub mod mcp_oauth;
74pub mod mcp_presets;
75pub mod mcp_progress;
76pub mod mcp_protocol;
77pub mod mcp_registry;
78pub mod mcp_sampling;
79pub mod mcp_server;
80pub mod metadata;
81pub mod module_artifact;
82pub mod observability;
83pub mod orchestration;
84pub mod personas;
85pub mod process_sandbox;
86pub mod profile;
87pub mod provenance;
88pub mod provider_catalog;
89pub mod receipts;
90pub mod record_filter;
91pub mod redact;
92pub mod run_events;
93pub mod runtime_context;
94pub(crate) mod runtime_guards;
95pub mod runtime_limits;
96pub mod runtime_paths;
97pub(crate) mod runtime_sqlite;
98pub mod schema;
99pub(crate) mod secret_patterns;
100pub mod secrets;
101pub mod security;
102pub mod session_bundle;
103pub mod session_timeline;
104pub mod sessions;
105pub(crate) mod shared_state;
106pub mod shells;
107pub mod skills;
108pub mod stdlib;
109pub mod stdlib_modules;
110pub mod step_runtime;
111pub mod store;
112pub(crate) mod synchronization;
113pub mod tenant;
114pub(crate) mod term;
115pub mod testbench;
116pub mod tool_annotations;
117pub mod tool_call_cancellations;
118pub mod tool_surface;
119pub mod tracing;
120pub mod triggers;
121pub mod trust_graph;
122pub(crate) mod url_encoding;
123pub mod user_dirs;
124
125/// Crate-wide deterministic clock mock used by stdlib time builtins, the
126/// trigger dispatcher, the cron scheduler, and Rust-side tests. Re-exports
127/// the long-lived implementation under `triggers::test_util::clock` so all
128/// callers go through one source of truth.
129pub mod clock_mock {
130    pub use crate::triggers::test_util::clock::{
131        active_mock_clock, advance, clear_overrides, install_override, instant_now, is_mocked,
132        now_ms, now_utc, sleep, ClockInstant, ClockOverrideGuard, MockClock,
133    };
134
135    /// Runtime audit for capabilities that observe real wall-clock or
136    /// monotonic time while a testbench mock is installed. See the module
137    /// docs for the full design.
138    pub mod leak_audit {
139        #[cfg(test)]
140        pub use crate::triggers::test_util::clock_leak::TEST_LOCK;
141        pub use crate::triggers::test_util::clock_leak::{
142            drain, instant_now, reset, snapshot, wall_now, ClockLeak,
143        };
144    }
145}
146
147pub mod typecheck;
148pub mod value;
149pub mod visible_text;
150mod vm;
151pub(crate) mod wait_for_graph;
152pub mod waitpoints;
153pub mod workspace_anchor;
154pub mod workspace_path;
155
156pub use actor_chain::{
157    ActorChain, ActorChainEntry, ActorChainError, Principal, ScopeAttenuationMode,
158    ScopeAttenuationPolicy, ScopeAttenuationViolation,
159};
160pub use builtin_id::BuiltinId;
161pub use call_budget::{
162    charge_mcp_call, charge_pg_query, install_mcp_call_budget, install_pg_query_budget,
163    McpCallBudgetGuard, PgQueryBudgetGuard,
164};
165pub use checkpoint::register_checkpoint_builtins;
166pub use chunk::*;
167pub use compiler::*;
168pub use connectors::{
169    active_connector_client, active_metrics_registry, clear_active_connector_clients,
170    clear_active_metrics_registry, connector_export_denied_builtin_reason,
171    connector_export_effect_class,
172    cron::{CatchupMode, CronConnector},
173    default_connector_export_policy,
174    harn_module::{
175        load_contract as load_harn_connector_contract, HarnConnector, HarnConnectorContract,
176    },
177    hmac::{verify_hmac_signed, SIGNATURE_VERIFY_AUDIT_TOPIC},
178    install_active_connector_clients, install_active_metrics_registry,
179    postprocess_normalized_event, ActivationHandle, ClientError, Connector, ConnectorClient,
180    ConnectorCtx, ConnectorError, ConnectorExportEffectClass, ConnectorHttpResponse,
181    ConnectorMetricsSnapshot, ConnectorNormalizeResult, ConnectorRegistry, GenericWebhookConnector,
182    HarnConnectorEffectPolicies, MetricsRegistry, PostNormalizeOutcome, ProviderPayloadSchema,
183    RateLimitConfig, RateLimiterFactory, RawInbound, StreamConnector, TriggerBinding, TriggerKind,
184    TriggerRegistry, WebhookSignatureVariant,
185};
186pub use corrections::{
187    append_correction_record, apply_corrections_to_policy, correction_query_filters_from_json,
188    correction_record_from_json, policy_with_corrections, query_correction_records,
189    CorrectionQueryFilters, CorrectionRecord, CorrectionScope, CORRECTIONS_TOPIC,
190    CORRECTION_EVENT_KIND, CORRECTION_SCHEMA_V0,
191};
192pub use harness::{
193    DenyEvent, Harness, HarnessCall, HarnessClock, HarnessCrypto, HarnessEnv, HarnessFs,
194    HarnessKind, HarnessLlm, HarnessNet, HarnessObs, HarnessProcess, HarnessRandom, HarnessSecrets,
195    HarnessStdio, HarnessSystem, HarnessTenant, HarnessTerm, MockAwareClock, MockHarnessBuilder,
196    VmHarness,
197};
198pub use harness_auth::{
199    current_auth_principal, enter_auth_principal, AuthPrincipal, AuthPrincipalScopeGuard,
200    MISSING_PRINCIPAL_MESSAGE,
201};
202pub use harness_net::{
203    bypass_enabled as net_policy_bypass_enabled, NetMatcher, NetPolicy, NetPolicyAudit,
204    NetPolicyDecision, NetPolicyDefault, NetPolicyRule, OnViolation, HARN_NET_POLICY_BYPASS_ENV,
205    NET_POLICY_AUDIT_TOPIC,
206};
207pub use harness_tenant::{
208    current_tenant_id, enter_tenant, TenantScopeGuard, MISSING_TENANT_MESSAGE,
209};
210pub use http::{register_http_builtins, reset_http_state};
211pub use llm::register_llm_builtins;
212pub use llm::trigger_predicate::TriggerPredicateBudget;
213pub use llm::{
214    current_agent_session_id, install_llm_cost_budget, install_llm_token_budget,
215    peek_llm_cost_budget, peek_llm_token_budget, register_session_end_hook, set_llm_cost_budget,
216    set_llm_token_budget, LlmBudgetGuard, LlmTokenBudgetGuard,
217};
218pub use mcp::{connect_mcp_server_from_json, connect_mcp_server_from_spec, register_mcp_builtins};
219pub use mcp_allowlist::{
220    build_catalog as build_mcp_catalog, catalog_for_request as mcp_catalog_for_request,
221    AdvertisedItem as McpAdvertisedItem, CatalogRequest as McpCatalogRequest, McpAllowlist,
222    McpAllowlistItem, McpCatalog, McpCatalogItem, McpCatalogServer, McpItemKind,
223    MCP_ALLOWLIST_SCHEMA_VERSION,
224};
225pub use mcp_card::{fetch_server_card, load_server_card_from_path, CardError};
226pub use mcp_host::{
227    cache_stats as mcp_host_cache_stats, set_allowlist as set_mcp_host_allowlist,
228    AllowlistDecision as McpHostAllowlistDecision, AllowlistGuard as McpHostAllowlistGuard,
229    BreakerState as McpHostBreakerState, CacheStats as McpHostCacheStats, McpHostStatus,
230    SpawnOptions as McpHostSpawnOptions, SupervisionPolicy as McpHostSupervisionPolicy,
231};
232pub use mcp_registry::{
233    active_handle as mcp_active_handle, ensure_active as mcp_ensure_active,
234    get_registration as mcp_get_registration, install_active as mcp_install_active,
235    is_registered as mcp_is_registered, register_servers as mcp_register_servers,
236    release as mcp_release, reset as mcp_reset_registry, snapshot_status as mcp_snapshot_status,
237    sweep_expired as mcp_sweep_expired, RegisteredMcpServer, RegistryStatus,
238};
239pub use mcp_server::{
240    take_mcp_serve_prompts, take_mcp_serve_registry, take_mcp_serve_resource_templates,
241    take_mcp_serve_resources, tool_registry_to_mcp_tools, McpServer,
242};
243pub use metadata::register_metadata_builtins;
244pub use observability::audit::{audit_events as audit_obs_events, AuditFinding, AuditFindingKind};
245pub use observability::request_id::{current_request_id, enter_request_id, RequestIdScopeGuard};
246pub use orchestration::{
247    benchmark_adapted_replay_pair, benchmark_replay_trace, build_replay_benchmark_report,
248    OpenCodeJsonlAdapter, ReplayBenchmarkCloudIngest, ReplayBenchmarkError,
249    ReplayBenchmarkFixtureReceipt, ReplayBenchmarkFixtureReport, ReplayBenchmarkMetrics,
250    ReplayBenchmarkReport, ReplayBenchmarkSuiteIdentity, ReplayBenchmarkSummary,
251    ReplayCategoryMetric, ReplayDebuggingProxyMetrics, ReplayRuntimeCostMetrics,
252    ReplayTraceAdapter, OPENCODE_JSONL_ADAPTER_ID, OPENCODE_JSONL_ADAPTER_SCHEMA_VERSION,
253    REPLAY_BENCHMARK_CLOUD_INGEST_KIND, REPLAY_BENCHMARK_REPORT_SCHEMA_VERSION,
254};
255pub use orchestration::{
256    canonicalize_run, first_divergence, run_replay_oracle_trace, ReplayAllowlistRule,
257    ReplayDivergence, ReplayExpectation, ReplayOracleError, ReplayOracleReport, ReplayOracleTrace,
258    ReplayTraceRun, ReplayTraceRunCounts, REPLAY_TRACE_SCHEMA_VERSION,
259};
260pub use orchestration::{
261    install_handoff_routes, snapshot_handoff_routes, HandoffRouteConfig,
262    HandoffRouteDecisionRecord, HandoffRouteTargetConfig,
263};
264pub use personas::{
265    disable_persona, fire_schedule as fire_persona_schedule, fire_trigger as fire_persona_trigger,
266    format_ms as format_persona_ms, now_ms as persona_now_ms, parse_rfc3339_ms as parse_persona_ms,
267    pause_persona, persona_status, record_persona_spend, register_persona_supervision_sink,
268    register_persona_value_sink, report_repair_worker_status, restore_persona_checkpoint,
269    resume_persona, PersonaAssignmentStatus, PersonaBudgetPolicy, PersonaBudgetStatus,
270    PersonaCheckpointAction, PersonaCheckpointRestoreOutcome, PersonaCheckpointRestoreRequest,
271    PersonaCheckpointResume, PersonaCheckpointUpdate, PersonaHandoffInboxItem, PersonaLease,
272    PersonaLifecycleState, PersonaQueuePositionUpdate, PersonaQueuedWork, PersonaReceiptUpdate,
273    PersonaRepairWorkerLifecycle, PersonaRepairWorkerStatusUpdate, PersonaRunCost,
274    PersonaRunReceipt, PersonaRuntimeBinding, PersonaStatus, PersonaSupervisionEvent,
275    PersonaSupervisionSink, PersonaSupervisionSinkRegistration, PersonaTriggerEnvelope,
276    PersonaValueEvent, PersonaValueEventKind, PersonaValueReceipt, PersonaValueSink,
277    PersonaValueSinkRegistration, StageDecl, StageExit, PERSONA_RUNTIME_TOPIC,
278};
279pub use provenance::{
280    build_signed_receipt, load_or_generate_agent_signing_key, verify_receipt, ProvenanceReceipt,
281    ReceiptBuildOptions, ReceiptVerificationReport,
282};
283pub use receipts::{
284    Receipt, ReceiptSink, ReceiptStatus, ReceiptValidationError, RedactingReceiptSink,
285    RedactionClass, RECEIPT_SCHEMA_ID, RECEIPT_SCHEMA_JSON, RECEIPT_SCHEMA_VERSION,
286};
287pub use record_filter::{normalize_record_filter_expression, CompiledRecordFilter};
288pub use runtime_limits::{
289    RuntimeLimitDescription, RuntimeLimitEntry, RuntimeLimits, RuntimeLimitsReport,
290    RUNTIME_LIMIT_DESCRIPTIONS,
291};
292pub use schema::json_to_vm_value;
293pub use sessions::{
294    CreateSession, ExpireSession, Session, SessionAttributes, SessionError, SessionStore,
295    TouchSession, SESSIONS_TOPIC,
296};
297pub use stdlib::hitl::{
298    append_hitl_response, ApprovalRequest, HitlHostResponse, HITL_APPROVALS_TOPIC,
299    HITL_DUAL_CONTROL_TOPIC, HITL_ESCALATIONS_TOPIC, HITL_QUESTIONS_TOPIC,
300};
301pub use stdlib::host::{clear_host_call_bridge, set_host_call_bridge, HostCallBridge};
302pub use stdlib::http_response::{
303    parse_envelope as parse_http_envelope, HttpEnvelope, HttpHeaderValue, WsUpgradeSpec,
304    HTTP_RESPONSE_TAG_KEY, HTTP_RESPONSE_TAG_VERSION,
305};
306#[cfg(feature = "postgres")]
307pub use stdlib::install_shared_pool_registry;
308pub use stdlib::io::{set_stdout_passthrough, take_stderr_buffer};
309pub use stdlib::long_running::cancel_handle as cancel_long_running_handle;
310pub use stdlib::observability::install_default_backend as install_obs_default_backend;
311pub use stdlib::secret_scan::{
312    append_secret_scan_audit, audit_secret_scan_active, scan_content as secret_scan_content,
313    SecretFinding, SECRET_SCAN_AUDIT_TOPIC,
314};
315pub use stdlib::template::{
316    lookup_prompt_consumers, lookup_prompt_span, prompt_render_indices, record_prompt_render_index,
317    PromptSourceSpan, PromptSpanKind,
318};
319pub use stdlib::waitpoint::{
320    process_waitpoint_resume_event, service_waitpoints_once, WAITPOINT_RESUME_TOPIC,
321};
322pub use stdlib::workflow_messages::{
323    workflow_pause_for_base, workflow_publish_query_for_base, workflow_query_for_base,
324    workflow_respond_update_for_base, workflow_resume_for_base, workflow_signal_for_base,
325    workflow_update_for_base, WorkflowMailboxState,
326};
327pub use stdlib::{
328    register_agent_stdlib, register_core_stdlib, register_io_stdlib, register_vm_stdlib,
329};
330pub use store::register_store_builtins;
331pub use tenant::{
332    tenant_event_topic_prefix, tenant_secret_namespace, tenant_topic, validate_tenant_id, ApiKeyId,
333    TenantApiKeyRecord, TenantBudget, TenantEventLog, TenantRecord, TenantRegistrySnapshot,
334    TenantResolutionError, TenantScope, TenantSecretProvider, TenantStatus, TenantStore,
335    TENANT_EVENT_TOPIC_PREFIX, TENANT_REGISTRY_DIR, TENANT_REGISTRY_FILE,
336    TENANT_SECRET_NAMESPACE_PREFIX,
337};
338pub use triggers::{
339    append_dispatch_cancel_request, begin_in_flight, binding_autonomy_budget_would_exceed,
340    binding_budget_would_exceed, binding_version_as_of, classify_trigger_dlq_error,
341    clear_dispatcher_state, clear_orchestrator_budget, clear_trigger_registry, drain,
342    dynamic_deregister, dynamic_register, expected_predicate_cost_usd_micros, finish_in_flight,
343    install_manifest_triggers, install_orchestrator_budget, install_provider_catalog,
344    micros_to_usd, note_autonomous_decision, note_orchestrator_budget_cost,
345    orchestrator_budget_would_exceed, parse_flow_control_duration, pause, pin_trigger_binding,
346    provider_metadata, record_predicate_cost_sample, redact_headers, register_provider_schema,
347    registered_provider_metadata, registered_provider_schema_names, reset_binding_budget_windows,
348    reset_provider_catalog, reset_provider_catalog_with, resolve_live_or_as_of,
349    resolve_live_trigger_binding, resolve_trigger_binding_as_of, resume,
350    run_trigger_harness_fixture, scheduler_in_flight_by_key, scheduler_ready_stats_by_key,
351    snapshot_dispatcher_stats, snapshot_orchestrator_budget, snapshot_trigger_bindings,
352    unpin_trigger_binding, usd_to_micros, worker_claims_topic_name, worker_job_topic_name,
353    worker_response_topic_name, ClaimedWorkerJob, DispatchCancelRequest, DispatchError,
354    DispatchOutcome, DispatchStatus, Dispatcher, DispatcherDrainReport, DispatcherStatsSnapshot,
355    FairnessKey, HeaderRedactionPolicy, InboxIndex, NotionPolledChangeEvent,
356    OrchestratorBudgetConfig, OrchestratorBudgetSnapshot, ProviderCatalog, ProviderCatalogError,
357    ProviderId, ProviderMetadata, ProviderOutboundMethod, ProviderPayload, ProviderRuntimeMetadata,
358    ProviderSchema, ProviderSecretRequirement, ReadyKeyStats, RecordedTriggerBinding, RetryPolicy,
359    SchedulableJob, SchedulerKeyStat, SchedulerPolicy, SchedulerSnapshot, SchedulerState,
360    SchedulerStrategy, SignatureStatus, SignatureVerificationMetadata, StreamEventPayload,
361    TenantId, TraceId, TriggerBatchConfig, TriggerBindingSnapshot, TriggerBindingSource,
362    TriggerBindingSpec, TriggerBudgetExhaustionStrategy, TriggerConcurrencyConfig,
363    TriggerDebounceConfig, TriggerDispatchOutcome, TriggerEvent, TriggerEventId,
364    TriggerExpressionSpec, TriggerFlowControlConfig, TriggerHandlerSpec, TriggerHarnessResult,
365    TriggerId, TriggerMetricsSnapshot, TriggerPredicateSpec, TriggerPriorityOrderConfig,
366    TriggerRateLimitConfig, TriggerRegistryError, TriggerRetryConfig, TriggerSingletonConfig,
367    TriggerState, TriggerThrottleConfig, WorkerQueue, WorkerQueueClaimHandle,
368    WorkerQueueEnqueueReceipt, WorkerQueueInspectSnapshot, WorkerQueueJob, WorkerQueueJobState,
369    WorkerQueuePriority, WorkerQueueResponseRecord, WorkerQueueState, WorkerQueueSummary,
370    DEFAULT_INBOX_RETENTION_DAYS, DEFAULT_STARVATION_AGE_MS, TRIGGERS_LIFECYCLE_TOPIC,
371    TRIGGER_ATTEMPTS_TOPIC, TRIGGER_CANCEL_REQUESTS_TOPIC, TRIGGER_DLQ_TOPIC,
372    TRIGGER_INBOX_CLAIMS_TOPIC, TRIGGER_INBOX_ENVELOPES_TOPIC, TRIGGER_INBOX_LEGACY_TOPIC,
373    TRIGGER_INBOX_OBSERVABILITY_TOPIC, TRIGGER_OPERATION_AUDIT_TOPIC, TRIGGER_OUTBOX_TOPIC,
374    TRIGGER_TEST_FIXTURES, WORKER_QUEUE_CATALOG_TOPIC,
375};
376pub use trust_graph::{
377    append_active_scope_attenuation_alert, append_active_trust_record,
378    append_scope_attenuation_alert, append_trust_record, export_trust_chain,
379    group_trust_records_by_trace, policy_for_agent, policy_for_autonomy_tier,
380    query_trust_graph_records, query_trust_records, resolve_agent_autonomy_tier,
381    summarize_trust_records, topic_for_agent, trust_score_for, verify_trust_chain, AutonomyTier,
382    TrustAgentSummary, TrustChainExport, TrustChainExportMetadata, TrustChainExportProducer,
383    TrustChainReport, TrustGraphRecord, TrustOutcome, TrustQueryFilters, TrustRecord,
384    TrustRecordActionKind, TrustScore, TrustTraceGroup, METADATA_KEY_ACTOR_CHAIN,
385    METADATA_KEY_ACTOR_CHAIN_ALERT, METADATA_KEY_EFFECTS_GRANT, METADATA_KEY_EFFECTS_USED,
386    METADATA_KEY_PARENT_RECORD_ID, OPENTRUSTGRAPH_ACCEPTED_SCHEMAS, OPENTRUSTGRAPH_CHAIN_SCHEMA_V0,
387    OPENTRUSTGRAPH_SCHEMA_V0, OPENTRUSTGRAPH_SCHEMA_V0_1, TRUST_ACTION_RELEASE,
388    TRUST_GRAPH_GLOBAL_TOPIC, TRUST_GRAPH_LEGACY_GLOBAL_TOPIC, TRUST_GRAPH_LEGACY_TOPIC_PREFIX,
389    TRUST_GRAPH_RECORDS_TOPIC, TRUST_GRAPH_TOPIC_PREFIX,
390};
391pub use value::*;
392pub use vm::*;
393
394#[cfg(feature = "vm-bench-internals")]
395#[doc(hidden)]
396pub mod bench_internals;
397
398/// Lex, parse, type-check, and compile source to bytecode in one call.
399/// Bails on the first type error. For callers that need diagnostics
400/// rather than early exit, use `harn_parser::check_source` directly
401/// and then call `Compiler::new().compile(&program)`.
402pub fn compile_source(source: &str) -> Result<Chunk, String> {
403    let program = harn_parser::check_source_strict(source).map_err(|e| e.to_string())?;
404    Compiler::new().compile(&program).map_err(|e| e.to_string())
405}
406
407/// Same as [`compile_source`] but compiles a specific named pipeline as
408/// the program entry point instead of the default-pipeline-or-first
409/// selection rule. Returns a runtime error when no pipeline with
410/// `pipeline_name` exists in the source.
411pub fn compile_source_named(source: &str, pipeline_name: &str) -> Result<Chunk, String> {
412    let program = harn_parser::check_source_strict(source).map_err(|e| e.to_string())?;
413    let has_pipeline = program.iter().any(|sn| {
414        let (_, inner) = harn_parser::peel_attributes(sn);
415        matches!(&inner.node, harn_parser::Node::Pipeline { name, .. } if name == pipeline_name)
416    });
417    if !has_pipeline {
418        return Err(format!("no pipeline named `{pipeline_name}` in source"));
419    }
420    Compiler::new()
421        .compile_named(&program, pipeline_name)
422        .map_err(|e| e.to_string())
423}
424
425pub fn json_schema_for_type_expr(type_expr: &harn_parser::TypeExpr) -> Option<serde_json::Value> {
426    let schema = compiler::Compiler::type_expr_to_schema_value(type_expr)?;
427    let json_schema = schema::schema_to_json_schema_value(&schema).ok()?;
428    Some(llm::vm_value_to_json(&json_schema))
429}
430
431pub fn json_schema_for_typed_params(params: &[harn_parser::TypedParam]) -> serde_json::Value {
432    let mut properties = serde_json::Map::new();
433    let mut required = Vec::new();
434
435    for param in params {
436        let param_schema = param
437            .type_expr
438            .as_ref()
439            .and_then(json_schema_for_type_expr)
440            .unwrap_or_else(|| serde_json::json!({}));
441        if param.default_value.is_none() {
442            required.push(serde_json::Value::String(param.name.clone()));
443        }
444        properties.insert(param.name.clone(), param_schema);
445    }
446
447    let mut schema = serde_json::Map::new();
448    schema.insert(
449        "type".to_string(),
450        serde_json::Value::String("object".to_string()),
451    );
452    schema.insert(
453        "properties".to_string(),
454        serde_json::Value::Object(properties),
455    );
456    if !required.is_empty() {
457        schema.insert("required".to_string(), serde_json::Value::Array(required));
458    }
459    serde_json::Value::Object(schema)
460}
461
462fn reset_llm_state_for_thread_reset() {
463    llm::reset_llm_state();
464    #[cfg(test)]
465    reset_thread_local_state_test_hooks::before_llm_global_reset();
466    // This full wipe is necessary between Harn programs to clear durable
467    // cooldowns that would otherwise stall a later run under a paused clock.
468    llm::reset_rate_limit_registry();
469    llm_config::clear_user_overrides();
470}
471
472#[cfg(test)]
473mod reset_thread_local_state_test_hooks {
474    use std::sync::{Arc, Mutex, OnceLock};
475
476    type Hook = Arc<dyn Fn() + Send + Sync + 'static>;
477
478    static BEFORE_LLM_GLOBAL_RESET: OnceLock<Mutex<Option<Hook>>> = OnceLock::new();
479
480    fn before_llm_global_reset_hook() -> &'static Mutex<Option<Hook>> {
481        BEFORE_LLM_GLOBAL_RESET.get_or_init(|| Mutex::new(None))
482    }
483
484    pub(crate) struct HookGuard;
485
486    impl Drop for HookGuard {
487        fn drop(&mut self) {
488            let mut hook = before_llm_global_reset_hook()
489                .lock()
490                .unwrap_or_else(std::sync::PoisonError::into_inner);
491            *hook = None;
492        }
493    }
494
495    pub(crate) fn install_before_llm_global_reset(hook: Hook) -> HookGuard {
496        let mut slot = before_llm_global_reset_hook()
497            .lock()
498            .unwrap_or_else(std::sync::PoisonError::into_inner);
499        *slot = Some(hook);
500        HookGuard
501    }
502
503    pub(crate) fn before_llm_global_reset() {
504        let hook = before_llm_global_reset_hook()
505            .lock()
506            .unwrap_or_else(std::sync::PoisonError::into_inner)
507            .clone();
508        if let Some(hook) = hook {
509            hook();
510        }
511    }
512}
513
514/// Reset all thread-local state that can leak between test runs.
515pub fn reset_thread_local_state() {
516    #[cfg(test)]
517    {
518        // `reset_thread_local_state` is also used by in-process unit tests. It
519        // clears process-global LLM config/rate-limit state, so share the same
520        // lock used by LLM env tests; otherwise a sibling reset can erase a
521        // parked rate-limit test's registry while the test still owns a permit.
522        let _guard = llm::env_guard();
523        reset_llm_state_for_thread_reset();
524    }
525    #[cfg(not(test))]
526    reset_llm_state_for_thread_reset();
527
528    http::reset_http_state();
529    channels::reset_channel_state();
530    event_log::reset_active_event_log();
531    egress::clear_explicit_egress_policy_requirement_for_host();
532    egress::clear_ssrf_guard_requirement_for_host();
533    stdlib::reset_stdlib_state();
534    connectors::clear_active_connector_clients();
535    orchestration::clear_runtime_hooks();
536    orchestration::clear_execution_policy_stacks();
537    orchestration::clear_command_policies();
538    orchestration::clear_pipeline_on_finish();
539    orchestration::reset_lifecycle_receipt_registry();
540    orchestration::agent_inbox::reset();
541    tool_call_cancellations::reset_registry();
542    redact::clear_policy_stack();
543    security::reset_thread_state();
544    triggers::clear_dispatcher_state();
545    triggers::clear_trigger_registry();
546    events::reset_event_sinks();
547    tracing::set_tracing_enabled(false);
548    tracing::reset_tracing();
549    agent_events::reset_all_sinks();
550    agent_sessions::reset_session_store();
551    mcp_registry::reset();
552    mcp_host::reset_for_tests();
553    call_budget::reset_call_budget_state();
554    clock_mock::leak_audit::reset();
555}
556
557#[cfg(test)]
558mod reset_leak_tests {
559    //! Regression coverage for harn#2660: process-/thread-global
560    //! registries that accumulated one entry per test because they were
561    //! never drained by `reset_thread_local_state`. Each case populates a
562    //! registry through its real entry point, runs the reset, and asserts
563    //! the registry is empty again.
564    use super::*;
565    use crate::value::VmValue;
566
567    #[test]
568    fn reset_drains_agent_inbox() {
569        orchestration::agent_inbox::reset();
570        orchestration::agent_inbox::push("sess-2660", "note", "leak", "test");
571        assert!(orchestration::agent_inbox::session_count() > 0);
572        reset_thread_local_state();
573        assert_eq!(
574            orchestration::agent_inbox::session_count(),
575            0,
576            "agent_inbox must be empty after reset"
577        );
578    }
579
580    #[test]
581    fn reset_drains_tool_call_cancellation_registry() {
582        tool_call_cancellations::reset_registry();
583        // Leak the guard so the entry survives until the reset runs —
584        // this mirrors a dispatch abandoned mid-flight.
585        let registered = tool_call_cancellations::register("sess-2660", "call-1", "tool");
586        if let Some((_handle, guard)) = registered {
587            std::mem::forget(guard);
588        }
589        assert!(tool_call_cancellations::registry_len() > 0);
590        reset_thread_local_state();
591        assert_eq!(
592            tool_call_cancellations::registry_len(),
593            0,
594            "tool-call cancellation registry must be empty after reset"
595        );
596    }
597
598    #[test]
599    fn reset_drains_routing_policy_registry() {
600        llm::routing::clear_policy_registry();
601        let mut config: crate::value::DictMap = crate::value::DictMap::new();
602        config.insert(
603            crate::value::intern_key("chain"),
604            VmValue::List(std::sync::Arc::new(vec![VmValue::String(
605                arcstr::ArcStr::from("mock:mock"),
606            )])),
607        );
608        llm::routing::build_routing_policy(&config).expect("intern a routing policy");
609        assert!(llm::routing::policy_registry_len() > 0);
610        reset_thread_local_state();
611        assert_eq!(
612            llm::routing::policy_registry_len(),
613            0,
614            "routing policy registry must be empty after reset"
615        );
616    }
617
618    #[test]
619    fn reset_holds_llm_env_guard_while_wiping_llm_globals() {
620        let observed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
621        let observed_hook = std::sync::Arc::clone(&observed);
622        let _hook = reset_thread_local_state_test_hooks::install_before_llm_global_reset(
623            std::sync::Arc::new(move || {
624                assert!(
625                    matches!(
626                        llm::env_lock().try_lock(),
627                        Err(std::sync::TryLockError::WouldBlock)
628                    ),
629                    "reset_thread_local_state must hold env_guard before wiping LLM globals"
630                );
631                observed_hook.store(true, std::sync::atomic::Ordering::SeqCst);
632            }),
633        );
634
635        reset_thread_local_state();
636        assert!(
637            observed.load(std::sync::atomic::Ordering::SeqCst),
638            "LLM global reset hook should have run"
639        );
640    }
641}
harn_vm/lib.rs

harn_vm/
lib.rs
