Python API

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

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

import noesis as ns

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

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)

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

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")

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,
)

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

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

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]

from noesis.io import last, list_runs

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

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

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

from noesis.exceptions import NoesisVeto

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

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

from noesis.governance import GovernanceFailurePolicy

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

from noesis.governance import GovernanceDecision

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

from noesis.governance import GovernanceResult

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

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}")

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)

from noesis.learn import LearnMode

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

from noesis.learn import LearnStatus

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

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
)

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
)

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