nined-agents 0.1.0

Python SDK for the 9D Labs runtime_v2 agent governance API

nined-agents

Python SDK for the 9D Labs runtime_v2 agent governance API.

Zero dependencies. Python 3.11+.

Install

pip install nined-agents

Quick start

from nined.agents import MemoryClient, PolicyDeniedError

client = MemoryClient(
    base_url="http://127.0.0.1:8082",
    api_key="your-api-key",
    world_id="my-world",
    workspace_id="ops-team",
    actor_id="agent-alpha",
    profile="builder",        # "builder" | "team" | "autonomy"
)

# Ingest a document
client.ingest([{
    "artifact_type": "document",
    "raw_payload": {"title": "Runbook", "content": "Restart pod if OOM."},
}])

# Retrieve relevant context
pack = client.context_pack("What do I do if pods are down?")

# Run a governed task (lock → work → complete → release, in one call)
client.start_task("task-001", title="Fix prod")
client.work_on_task("task-001", "restart_deployment")

Profiles

Profile	Capabilities
builder	memory + basic task ops
team	+ handoffs, locks, replay
autonomy	+ diffs, alerts (default)

All methods

Memory

• ingest(artifacts) — store documents/artifacts
• context_pack(query, *, max_tokens, profile) — retrieve context

Tasks

• start_task(task_id, *, title, metadata)
• attempt_action(task_id, action_name, *, decision_type, reason, ...)
• complete_task(task_id)
• fail_task(task_id, *, reason)

Coordination

• acquire_lock(task_id, *, ttl_seconds)
• release_lock(task_id)
• handoff(task_id, *, to_actor_id, reason)

Convenience (recommended)

• work_on_task(task_id, action_name, *, reason, complete) — full lock→work→complete→release
• delegate_task(task_id, to_actor_id, *, reason) — release + handoff
• escalate_task(task_id, *, reason) — signals need for higher authority

State

• get_task_state(task_id) — includes allowed_actions + protocol_hint
• get_timeline(task_id) — append-only event history

Ops

• replay(pack_hash) — verify determinism
• get_receipts(*, limit, offset)
• upsert_policy_bundle(bundle_version, *, role_capabilities, constraints)
• get_policy_bundle()

Errors

from nined.agents import PolicyDeniedError

try:
    client.complete_task("t1")
except PolicyDeniedError as exc:
    print(exc.reason_code)           # e.g. "policy_denied_invalid_state_transition"
    print(exc.recommended_action)    # hint string
    print(exc.recovery)              # {"recovery_action": "get_task_state", ...}

Exception	When
PolicyDeniedError	Runtime denied the action (state violation, missing approval, etc.)
ConflictError	Lock contention or stale handoff
AuthorizationError	Bad API key or missing capabilities
MemoryAPIError	Any other API error (base class)

Runtime-guided behavior

get_task_state() always returns allowed_actions and protocol_hint:

state = client.get_task_state("task-001")
print(state["allowed_actions"])  # ["attempt_action", "complete_task", "release_lock"]
print(state["protocol_hint"])    # "You own this task and hold the lock. Do your work..."

Agents can use these to drive a self-correcting action loop without hardcoding state logic.

TLA+ verification

The specs/TaskFSM.tla model verifies four invariants over all reachable states:

• Terminal states are irreversible
• Lock is exclusive (at most one holder)
• Actions that mutate task state require holding the lock
• Only the owner can act on a task

java -XX:+UseParallelGC -jar tla2tools.jar \
  -config specs/TaskFSM.cfg -workers 4 -nowarning -deadlock specs/TaskFSM.tla