Python API - Noēsis

Install from source while the PyPI release is pending:

git clone https://github.com/saraeloop/noesis.git
cd noesis
uv tool install .
# or: pipx install .

When PyPI is live, use uv add noesis or pip install noesis. Import as import noesis as ns.

Core functions

ns.run()

Execute a baseline episode using the current session.

episode_id = ns.run(
    task: str,
    *,
    seed: int = 0,
    intuition: bool | Intuition | None = True,
    tags: dict[str, object] | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
    verify: VerifySpec | Sequence[VerifySpec] | None = None,
) -> str

string

required

Task or goal for the episode.

int

default:"0"

Seed for reproducibility.

bool | Intuition | None

default:"True"

True enables the default intuition policy, False disables it, or pass an Intuition implementation.

dict

Metadata tags attached to the episode.

RuntimeContext

default:"None"

Optional runtime context. If provided, execution bypasses the default session.

str | Path | None

default:"None"

Workspace root to snapshot for verification.

VerifySpec | Sequence[VerifySpec] | None

default:"None"

Verification assertions to evaluate against the workspace.

Returns: Episode ID (e.g., "ep_01JH6Z2V9Q2K6Y6N0QZ7K2QW8C") Example:

import noesis as ns

episode_id = ns.run("Draft release notes")
episode_id = ns.run("Draft release notes", intuition=False, tags={"env": "staging"})

Runtime context example:

import noesis as ns
from noesis.context import create_runtime_context

runtime = create_runtime_context()
episode_id = ns.run("Summarize this incident", context=runtime)
summary = ns.summary.read(episode_id, context=runtime)

context is a RuntimeContext (a ports container). For metadata, prefer tags={...}.

ns.solve()

Execute an episode through a specific adapter/graph.

episode_id = ns.solve(
    task: str,
    *,
    using: GraphSource,  # import path, callable, or adapter
    seed: int = 0,
    intuition: bool | Intuition | None = True,
    tags: dict[str, object] | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
    verify: VerifySpec | Sequence[VerifySpec] | None = None,
) -> str

GraphSource

required

Adapter name/import path or callable to execute the task.

Example:

import noesis as ns

# Callable adapter
def to_upper(task: str) -> dict:
    return {"result": task.upper()}

episode_id = ns.solve("process this", using=to_upper)
episode_id = ns.solve("process this", using="my.module:adapter_fn")

Run lifecycle APIs

Use these APIs for approval gates and same-run continuation. Simple mental model:

interrupt and checkpoint pause a run with audit evidence.
resume emits lifecycle evidence only (run.resume).
resume_run emits run.resume and continues execution.

event_id = ns.interrupt(
    episode_id: str,
    *,
    reason: str | None = None,
    caused_by: str | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
) -> str

checkpoint = ns.checkpoint(
    episode_id: str,
    *,
    caused_by: str | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
) -> dict[str, object]

resume_event_id = ns.resume(
    episode_id: str,
    *,
    checkpoint_id: str,
    caused_by: str | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
) -> str

episode_id = ns.resume_run(
    episode_id: str,
    *,
    checkpoint_id: str,
    using: GraphSource | None = None,
    caused_by: str | None = None,
    context: Any | None = None,
    workspace: str | Path | None = None,
    verify: VerifySpec | Sequence[VerifySpec] | None = None,
) -> str

Lifecycle flow (easy)

import noesis as ns

# 1) Optional: enable pause-on-veto in enforce mode
ns.set(
    governance_mode="enforce",
    governance_pause_on_veto=True,
)

# 2) Start a run
episode_id = ns.solve("Danger operation: delete production database", using=my_graph)

# 3) Manual pause (if your approval system decides to pause)
interrupt_id = ns.interrupt(episode_id, reason="Awaiting human approval")
checkpoint = ns.checkpoint(episode_id, caused_by=interrupt_id)

# 4a) Emit resume evidence only (no execution continuation)
resume_event_id = ns.resume(episode_id, checkpoint_id=checkpoint["checkpoint_id"])

# 4b) Continue execution from checkpoint on the same run ID
episode_id = ns.resume_run(
    episode_id,
    checkpoint_id=checkpoint["checkpoint_id"],
    using=my_graph,  # required for non-minimal runs
)

Determinism scope:

Noēsis enforces immutable, append-only artifacts and replay discipline.
External tool/LLM outputs may still vary unless your environment captures/freezes them.

Continuation contract:

Same run ID.
Append-only artifacts preserved.
Resume continues post-plan by default (no replan) with anchor validation.

Adapter continuity:

resume_run(..., using=...) must match the persisted adapter from checkpoint/state.
For minimal runs (ns.run(...)), omitting using is allowed.

Common failure modes

RunSealedError: lifecycle writes and resume attempts are rejected once final.json seals the run.
CheckpointNotFoundError: resume/resume_run reference a checkpoint that does not exist.
MissingCausalParentError: checkpoint/interrupt cannot anchor to a causal parent event.
CheckpointConsistencyError: checkpoint anchor (event_offset, last_event_id, state_hash) no longer matches artifacts.
RunLifecycleTransitionError: lifecycle mutation violates the run state-machine contract.
ResumeAdapterRequiredError: resume_run requires explicit using for non-minimal checkpoints.
ResumeAdapterMismatchError: resume_run adapter does not match checkpoint adapter contract.

All errors above are defined in noesis.domain.run_lifecycle.

Verification helpers

Use these helpers to build verification specs for verify=....

verify = [
    ns.file_exists("config.yaml"),
    ns.file_contains("config.yaml", "enabled: true"),
    ns.only_modified(["config.yaml"]),
    ns.no_modifications(),
]

episode_id = ns.solve(
    "Update config",
    using="my.module:adapter_fn",
    workspace=".",
    verify=verify,
)

ns.summary.read()

Load the summary for an episode.

summary = ns.summary.read(episode_id: str, *, context: Any | None = None) -> dict

Returns: Summary dictionary (task, metrics, flags, manifest, etc.).

ns.events.read()

Load the event timeline for an episode.

events = ns.events.read(
    episode_id: str,
    *,
    stream: bool = False,
    context: Any | None = None,
) -> Iterable[dict]

Set stream=True to iterate lazily. Integrity behavior:

Raises noesis.trace.events.EventLogIntegrityError when events.jsonl contains invalid UTF-8, malformed JSON, or a non-object record.
Uses fail-closed semantics: corrupted lines are not skipped.

import noesis as ns
from noesis.trace.events import EventLogIntegrityError

try:
    for event in ns.events.read("ep_01JH6Z2V9Q2K6Y6N0QZ7K2QW8C", stream=True):
        print(event["phase"])
except EventLogIntegrityError as exc:
    print(exc.corruption.path, exc.corruption.line_number, exc.corruption.reason)
    raise

noesis.io.list_runs()

List recent episodes (newest first).

from noesis.io import list_runs

episodes = list_runs(
    limit: int = 50,
    since: str | None = None,
    *,
    context: Any | None = None,
    strict_manifest: bool = False,
) -> list[dict]

Each row includes episode_id, task, started_at, flags, success, manifest, and manifest_status (when strict_manifest=True).

In v1.0.0, ns.list_runs() exists as a deprecated legacy alias. Prefer noesis.io.list_runs().

noesis.io.last()

Get the most recent episode ID.

from noesis.io import last, list_runs

episode_id = last(*, context: Any | None = None) -> str | None

ns.set() / ns.get()

Update or read the current configuration snapshot.

ns.set(runs_dir=".noesis/episodes", planner_mode="minimal", direction_min_confidence=0.7)
config = ns.get()  # returns a mapping of current config values

runs_dir points to the episodes root directory (for example, .noesis/episodes). Common keys: runs_dir, planner_mode (meta/minimal), direction_min_confidence, governance_mode (off/audit/enforce), governance_failure_policy, governance_timeout_ms (reserved/unused), governance_pause_on_veto, policy_aliases, learn_home, learn_mode, learn_auto_apply_min_confidence, learn_auto_apply_min_successes, intuition_mode, timeout_sec, prompt_provenance_enabled, prompt_provenance_mode, agents (reserved/unused), tasks (reserved/unused).

Intuition and policies

DirectedIntuition

Base class for policies that can emit hints, interventions, or vetoes.

from noesis.direction import DirectedIntuition


class SafetyPolicy(DirectedIntuition):
    __version__ = "1.0"

    def advise(self, state: dict) -> IntuitionEvent | None:
        task = str(state.get("task", "")).lower()
        if "delete" in task:
            return self.veto(
                advice="Blocked: delete operation",
                rationale="Delete requires manual approval",
                target="plan",
            )
        return None

Helper methods:

hint(advice, confidence=0.5, rationale=None, evidence_ids=None, target="input", scope="episode")
intervene(advice, patch, confidence=0.6, rationale=None, evidence_ids=None, target="input", scope="episode")
veto(advice, confidence=0.8, rationale=None, evidence_ids=None, target="plan", scope="episode")

IntuitionEvent (schema)

Fields include kind, advice, confidence, policy_id, policy_version, policy_kind, applied, rationale, evidence_ids, patch, target, scope, and blocking (plus schema_version).

NoesisVeto

Raised when a policy vetoes an episode.

from noesis.exceptions import NoesisVeto

try:
    ns.run("DELETE * FROM users", intuition=True)
except NoesisVeto as e:
    print(f"Vetoed: {e}")

Governance

Pre-act governance evaluates proposed actions before execution. Configure via ns.set(governance_mode=...).

GovernanceMode

from noesis.governance import GovernanceMode

GovernanceMode.OFF      # Governance disabled (default)
GovernanceMode.AUDIT    # Record decisions; never blocks execution
GovernanceMode.ENFORCE  # Veto terminates the episode before Act

GovernanceFailurePolicy

from noesis.governance import GovernanceFailurePolicy

GovernanceFailurePolicy.FAIL_OPEN    # On error, allow action
GovernanceFailurePolicy.FAIL_CLOSED  # On error, treat as veto

Default depends on mode: audit → fail_open, enforce → fail_closed.

GovernanceDecision

from noesis.governance import GovernanceDecision

GovernanceDecision.ALLOW  # Action proceeds
GovernanceDecision.AUDIT  # Action proceeds, flagged for review
GovernanceDecision.VETO   # Action blocked

GovernanceResult

Immutable result from governance evaluation.

from noesis.governance import GovernanceResult

# Fields: decision, rule_id, score, message, policy_id, policy_version,
#         policy_kind, mode, failure_policy, enforced, error, details

Custom governors

Custom governor injection is not part of the v1.0.0 runtime/CLI execution surface. Governance is configured via ns.set(governance_mode=..., governance_failure_policy=...) and uses the built-in pre-act governor.

Governed side effects (pre-act gating)

ns.governed_act(...) is the operating-system boundary for side effects. It executes through the same canonical runtime boundary used by ns.run(...) and ns.solve(...) (same finalization and sealing rules). Event paths:

allow/audit: action_candidate → governance → act
enforce veto (governance_pause_on_veto=False): action_candidate → governance → terminate (no act)
enforce veto with pause enabled (governance_pause_on_veto=True): action_candidate → governance → run.interrupt → run.checkpoint (no act, no terminate)

Artifact constraints:

terminal outcomes (allow/audit/veto terminate): run seals with final.json and manifest.json
paused-on-veto outcomes: run stays unsealed (no final.json / manifest.json yet) until continuation/termination

import noesis as ns
from noesis.exceptions import NoesisVeto

def run_shell(*, command: str, cwd: str | None = None, timeout_ms: int | None = None):
    return {"stdout": "ok", "stderr": "", "exit_code": 0, "command": command}

ns.set(shell_executor=run_shell)

try:
    result = ns.governed_act(
        goal="List repository files",
        kind="shell",
        payload={
            "command": "ls -a",
            "cwd": ".",
            "timeout_ms": 2000,
        },
    )
    print(result)
except NoesisVeto as veto:
    # Raised only when governance is enforcing and the action is vetoed.
    print(f"Blocked by governance: {veto.advice}")

Supported kind values:

shell: requires ns.set(shell_executor=...)
adapter: requires ns.set(adapter_executor=...)

If the matching executor is not configured, ns.governed_act(...) raises ValueError before actuation.

Troubleshooting `ns.governed_act(...)`

Symptom	Likely cause	Fix
`ValueError: shell executor is not configured`	called with `kind="shell"` before registering an executor	register one with `ns.set(shell_executor=...)`
`ValueError: adapter executor is not configured`	called with `kind="adapter"` before registering an executor	register one with `ns.set(adapter_executor=...)`
`ValueError: unsupported action kind`	`kind` is not one of `shell` or `adapter`	use a supported kind and route custom logic through your registered executor
run pauses and remains unsealed	enforce-mode veto with `governance_pause_on_veto=True`	resume with `ns.resume_run(...)` after approval, or terminate explicitly

Internal tool invocation use cases (adapter implementers)

Noesis also exposes an internal application contract for protocol adapters:

from noesis.usecases.tool_invocation import (
    ToolInvocationInput,
    prepare_tool_invocation,
    execute_prepared_tool_invocation,
)

Use this contract when you need explicit prepare/approve/execute boundaries for side-effecting tools. The canonical identity key is run_id + draft_id. Execution-time failures are surfaced as typed errors from noesis.domain.tool_contract:

PreparedToolInvocationNotFoundError
ApprovalDecisionRequiredError
ApprovalDecisionBindingError

For full workflow, event ordering, and troubleshooting guidance, see Integrate adapters.

Session management

Use sessions when you need isolated configuration, explicit lifecycle control, or registered ports.

from noesis import SessionBuilder

session = SessionBuilder.from_env().build()
ep = session.run("Process customer request")

# Read artifacts using the session's RuntimeContext
import noesis as ns
summary = ns.summary.read(ep, context=session.context)

SessionBuilder reads config from env/TOML; you can also inject ports before building. Within a session, run/solve behave like the module-level helpers but share the session’s config and runtime context.

Module facades

ns.summary.read(episode_id, context=None): read summary.json.
ns.events.read(episode_id, stream=False, context=None): iterate events; pass stream=True to lazily consume.
ns.context: helpers for building runtime contexts and attaching ports (advanced use — see the “Add a memory port” guide).
ns.learn: learning signal emission and proposal management (see Learning section below).

Learning

The learning subsystem records proposals from episode outcomes for policy improvement.

LearnMode

from noesis.learn import LearnMode

LearnMode.OFF     # Disable learning
LearnMode.RECORD  # Record proposals only (default)
LearnMode.APPLY   # Auto-apply approved proposals

Configure via ns.set(learn_mode="record").

LearnStatus

from noesis.learn import LearnStatus

LearnStatus.PENDING   # Awaiting review
LearnStatus.APPROVED  # Approved for application
LearnStatus.REJECTED  # Rejected
LearnStatus.APPLIED   # Already applied

LearnProposal

Dataclass for learning signals. Contains kind, payload, confidence, status, and metadata.

In v1.0.0, learning proposals are emitted automatically during summary finalization when learn_mode is enabled. When proposals are generated, they are written to learn.jsonl for the episode and tracked under learn_home.

Helper functions

from noesis.learn import (
    build_learn_payload,      # Build structured learn payloads
    persist_episode_learning, # Write learn.jsonl
    load_policy_snapshot,     # Read current policy state
    update_policy_snapshot,   # Update policy state
    derive_target_key,        # Compute target key from proposal
    summarise_learn_kinds,    # Summarize proposal kinds
)

Episode index

EpisodeIndex

Manage an on-disk episode manifest (and optional FAISS similarity index).

from noesis.episode import EpisodeIndex

store = EpisodeIndex(
    root="./.noesis/episodes/_episodes",
    ttl_days=14,
    enable_faiss=False,  # set True if faiss + numpy installed and you provide embeddings
)

Core methods:

append(episode_id, summary_path, state_path, status, task, using, provenance=None, embedding=None)
iter(include_expired=False) → iterator of EpisodeRecord
search(embedding, k=5) → similarity matches (empty if FAISS disabled)
vacuum() → prune expired records

Type definitions

IntuitionEvent

Returned by policy methods. See schema above for fields.

Determinism utilities

For reproducible testing and replay, Noesis exports deterministic clock and RNG utilities:

from datetime import datetime, timezone
from noesis import DeterministicClock, DeterministicRNG

# Fixed-time clock for testing
clock = DeterministicClock(start_at=datetime(2025, 1, 1, tzinfo=timezone.utc))
now = clock.now()  # Always returns the same timestamp

# Seeded RNG for reproducibility
rng = DeterministicRNG(seed=42)
value = rng.bytes(8)  # Deterministic bytes

These are used internally by the replay gate to ensure artifact determinism. Exposed for custom test harnesses and advanced integrations.

Environment variables

Variable	Description
`NOESIS_RUNS_DIR`	Artifact storage directory
`NOESIS_PLANNER`	Planner mode (`meta`/`minimal`)
`NOESIS_DIRECTION_MIN_CONFIDENCE`	Direction minimum confidence
`NOESIS_GOVERNANCE_MODE`	Governance mode (`off`/`audit`/`enforce`)
`NOESIS_GOVERNANCE_FAILURE_POLICY`	Failure policy (`fail_open`/`fail_closed`)
`NOESIS_GOVERNANCE_TIMEOUT_MS`	Reserved/unused in v1.0.0 (parsed into config)
`NOESIS_INTUITION_MODE`	Intuition mode
`NOESIS_TIMEOUT_SEC`	Default timeout (seconds)
`NOESIS_LEARN_HOME`	Learning artifacts directory
`NOESIS_LEARN_MODE`	Learning mode
`NOESIS_LEARN_AUTO_APPLY_MIN_SUCCESSES`	Minimum successes before auto-apply
`NOESIS_LEARN_AUTO_APPLY_MIN_CONFIDENCE`	Confidence threshold for auto-apply
`NOESIS_PROMPT_PROVENANCE_ENABLED`	Enable prompt provenance (`true`/`false`)
`NOESIS_PROMPT_PROVENANCE_MODE`	Prompt provenance mode
`NOESIS_AGENTS`	Reserved/unused in v1.0.0 (parsed into config)
`NOESIS_TASKS`	Reserved/unused in v1.0.0 (parsed into config)

​Core functions

​ns.run()

​ns.solve()

​Run lifecycle APIs

​Lifecycle flow (easy)

​Common failure modes

​Verification helpers

​ns.summary.read()

​ns.events.read()

​noesis.io.list_runs()

​noesis.io.last()

​ns.set() / ns.get()

​Intuition and policies

​DirectedIntuition

​IntuitionEvent (schema)

​NoesisVeto

​Governance

​GovernanceMode

​GovernanceFailurePolicy

​GovernanceDecision

​GovernanceResult

​Custom governors

​Governed side effects (pre-act gating)

​Troubleshooting ns.governed_act(...)

​Internal tool invocation use cases (adapter implementers)

​Session management

​Module facades

​Learning

​LearnMode

​LearnStatus

​LearnProposal

​Helper functions

​Episode index

​EpisodeIndex

​Type definitions

​IntuitionEvent

​Determinism utilities

​Environment variables

​Next steps

CLI reference

Write policies

Core functions

ns.run()

ns.solve()

Run lifecycle APIs

Lifecycle flow (easy)

Common failure modes

Verification helpers

ns.summary.read()

ns.events.read()

noesis.io.list_runs()

noesis.io.last()

ns.set() / ns.get()

Intuition and policies

DirectedIntuition

IntuitionEvent (schema)

NoesisVeto

Governance

GovernanceMode

GovernanceFailurePolicy

GovernanceDecision

GovernanceResult

Custom governors

Governed side effects (pre-act gating)

Troubleshooting `ns.governed_act(...)`

Internal tool invocation use cases (adapter implementers)

Session management

Module facades

Learning

LearnMode

LearnStatus

LearnProposal

Helper functions

Episode index

EpisodeIndex

Type definitions

IntuitionEvent

Determinism utilities

Environment variables

Next steps