Rivus Agent

A small local agent daemon core for experimenting with a Pi-backed agent loop and Feishu IM integration.

Documentation: English | 简体中文

The first milestone is intentionally narrow: make the agent loop and agent harness layer pleasant to use before adding transport adapters. The core is built around DDD-style domain events and Effect-powered use cases.

Current Status

• TypeScript ESM package.
• Public Plugin SDK with atomic Plugin registration, immutable Catalog snapshots, deterministic Agent Deployment resolution, an explicit Required/Optional Plugin Loader, safe Node module resolution from a Deployment Root, and a manifest-driven Multi-agent Deployment Daemon. Concrete Agent products remain external packages.
• Host-bound Workspace Instructions API and built-in AGENTS.md provider with deterministic root-to-target discovery, immutable source digests, nested mutation refresh diagnostics, UTF-8/path/symlink checks, and whole-source byte budgeting. The deployment example injects the root View into Pi while Tool-target refresh remains enforced by the provider boundary.
• Host-issued invocation authority and a call-time Tool Broker that checks exact grants, pinned versions, online revocation, stable operation IDs, durable side-effect fencing, and one-time approval receipts bound to canonical input digests. The Pi Tool Proxy and Feishu Human Interaction gateway are wired in the deployment example.
• Multi-agent Host primitives with trusted Endpoint bindings, stable Runtime Generation/Instance identities, lazy per-instance Runtime pooling, isolated active-run gates, and restart restoration from persisted registry records.
• Scoped Agent Memory with a single-writer append-only JSONL adapter, explicit Profile/Deployment grants, Host-bound tenant/subject/conversation/project/audience identity, and a provider-neutral rivus_memory Tool wired into the Pi manifest Runtime. Agent-private proposals survive process replacement; group invocations may access only Conversation and confirmed Project Memory, and Shared Profile reads expose only confirmed records. Bound Project Memory is recalled automatically with deterministic limits; durable confirmation UX, typed Shared Profile fields, full provenance/retention metadata, and durable Compaction are still separate work.
• Manifest-driven fixed-time Agent Automations with daily or weekday-only schedules, stable Mandate/Tick identity, per-template Tool grant narrowing, isolated Runtime and invocation authority, a durable single-writer JSON Tick adapter that records the actual Agent Run, generated-before-delivery recovery, and idempotent proactive Feishu interactive-card delivery. The first Markdown heading becomes the card title while the remaining Markdown stays scannable in the card body. Revisioned Plugin State and the general Delivery Outbox remain composable domain primitives.
• Opt-in background Agent Sessions (manifest.backgroundSessions): the model decides from ordinary natural language that work should continue in the background, calls the generic background.start Host Tool, and the foreground Run and card end promptly. A deployment-owned supervisor runs bounded steps on an isolated Runtime lane per Agent with the exact Agent/Profile/Project Space/Skills/Memory policy and a grant narrowed to exclude background.start; sessions wait durably (background.wait), wake, receive user input once (background.send), stop safely, and survive daemon restart on the same stable session and Pi transcript. Progress and one terminal result are delivered idempotently to the trusted source conversation. ACP runtimes reach the same Host Tools through a built-in MCP bridge (@rivus/agent/mcp), so Pi and ACP backends share the identical domain, supervisor, persistence, status, and delivery semantics. The Runtime contains no business scenario, meeting, or job code — no pre-model intent router exists.
• Peer Delegation and Subagent coordination primitives with explicit edges, hop/budget limits, monotonic ToolGrantSet intersection, isolated child authority, denied ambient sender/private-memory/Shell capabilities, and parent cancellation cascading.
• @rivus/agent/testing Conformance TestKit with a reusable Fake Plugin and checks for registration/reference failures, ToolGrantSet expansion, and lifecycle disposal leaks.
• Effect-based AgentHarness use case with result-oriented prompt(), turn-oriented promptTurn(), event-oriented stream(), text-oriented streamText(), event-handling promptWithEvents(), active-run/run/session state and transcript lookup APIs, and a default runtime factory.
• Session-first AgentRuntime convenience facade that keeps the Effect-first harness and Promise client together with a default main session.
• Pure domain events and reducer for agent run state, including separate assistant text and thinking projections.
• Optional Effect-based domain event sinks on AgentHarness, plus a JSONL event log and run/session history/transcript replay.
• Production runtime helpers for system time and UUID run ids, restored run-state seeding, plus deterministic testing fakes.
• A small AgentLoop port that future Pi SDK and RPC adapters can implement, including cancellation signals plus same-session state, transcript, and conversation-message handoff, plus helpers for async event generators, simple text generators, and string shorthand text deltas.
• A Pi SDK loop adapter that maps AgentSession.subscribe() text, thinking, and tool events into daemon loop events, aborts active Pi work when the loop input signal is aborted, and accepts SDK args plus hook/RPC-shaped input tool payloads.
• Feishu message/card-action intake helpers, callback toast plus card-update responses, exact run cancellation commands, trusted-operator Human Interaction resolution, stable session-key construction, and stream-action projection.
• A Pi session registry that reuses sessions by stable SessionKey.
• A thin Feishu daemon use case that wires intake, dedupe, harness execution, and stream actions together.
• In-memory and JSON file-backed Feishu CardKit target registries for binding agent runs to card element targets.
• CardKit target preparation helpers that create a streaming card entity, reply to the source message, and bind { cardId, elementId } before streaming starts.
• A CardKit publisher port with per-run sequence handling, text-update coalescing, periodic flushing, and a simple 5 QPS rate-limit wrapper.
• A fast-ack Feishu message queue and worker that separate event acceptance from agent execution and keep failed drains pending for retry.
• A small Feishu agent runtime factory that assembles the harness, queue, worker, and event handlers.
• A Feishu WebSocket daemon adapter that starts an official-SDK-shaped WSClient with the runtime handlers.
• A background Feishu worker loop that drains accepted messages outside the WebSocket acknowledgement callback.
• A small rivus CLI with conflict-safe project initialization, side-effect-free local readiness diagnostics, bootstrap loading, daemon startup, and SIGINT/SIGTERM shutdown.
• A configured daemon bootstrap helper that wires Feishu WebSocket, CardKit target preparation, CardKit publishing, event logging, history replay, worker draining, and managed process lifecycle.
• A daemon status reporter that combines active harness run details, process/transport/worker/queue state, and restored run/session history.
• A small CardKit OpenAPI client for streaming text updates and replacing the card with completed, failed, or cancelled terminal projections.
• Durable per-Endpoint Human Interaction services with JSONL snapshots, separate Tool Approval and User Decision state machines, allowlisted Feishu actors, expiring requests, and exact one-time Tool Broker consumption.
• Feishu tenant access token provider, JSON fetch adapter, configured CardKit target preparation, configured CardKit publisher, and receive-message handler map for SDK integration.
• Tests covering prompt execution, event subscription, durable event sinks, busy-run rejection, run/session state lookup, state snapshots, and loop failure events.

Production SDK bootstrap files can stay thin by creating SDK clients and passing them into the configured bootstrap helper. See examples/pi-feishu.bootstrap.ts for a concrete Pi SDK plus Feishu Node SDK long-connection bootstrap.

Package Installation

Published releases require Node 24 LTS and install from the public npm registry:

npm install @rivus/agent
npx rivus --help

Create a standalone one-Agent project without cloning this repository:

npm install --global @rivus/agent
rivus init my-rivus-agent
cd my-rivus-agent
npm install
cp .env.example .env.local
# Fill the Feishu and model-provider values in .env.local.
npm run doctor
npm start

init refuses to overwrite any generated path. doctor checks Node, project files, local dependencies, the deployment manifest, and enabled Endpoint credentials without importing Plugin code or contacting external services. See the local deployment runbook for foreground verification and the generated systemd/launchd templates.

The Pi plus Feishu example is application-edge code: install @earendil-works/pi-coding-agent and @larksuiteoapi/node-sdk in the daemon app that uses it. The @rivus/agent core package keeps those SDKs out of its runtime dependency graph.

npm install @earendil-works/pi-coding-agent@^0.80.6 @larksuiteoapi/node-sdk@^1.70.0

Quick Start

curl -fsSL https://vite.plus | bash
vp install
vp run check

vp run check is the complete local and CI gate. Vite+ runs Oxfmt, type-aware Oxlint with TypeScript diagnostics, Vitest, and library packaging; the gate then packs and installs the npm archive in a clean temporary consumer, exercises both public exports and the CLI, runs the high-severity dependency audit, and checks documentation and repository safety. Use vp check --fix for deterministic formatting and safe lint fixes, then review the diff before committing. The project runtime is pinned in .node-version.

npm releases are produced only by the Publish npm GitHub Actions workflow after a GitHub Release is published for an exact v<package version> tag on main. The workflow uses npm Trusted Publishing through OIDC and contains no long-lived registry token. See the npm release runbook for the one-time first-package bootstrap and normal release procedure.

For a manifest-driven multi-agent deployment, keep rivus.config.json, its lockfile, and installed Plugin packages in one deployment project. createConfiguredRivusDeploymentDaemon() validates the manifest, resolves only declared modules from that directory, disables deployments belonging to failed Optional Plugins, and blocks startup when a Required Plugin fails:

{
  "plugins": [
    {
      "id": "research-agent",
      "module": "research-agent/rivus-plugin",
      "required": true
    }
  ],
  "defaultAgentId": "researcher",
  "defaultEndpointId": "research-feishu",
  "agents": [
    {
      "agentId": "researcher",
      "pluginId": "research-agent",
      "profileId": "researcher",
      "endpointIds": ["research-feishu"],
      "memory": { "scopes": ["agent-private"], "tool": true },
      "tools": { "allow": ["research-agent/read"] }
    }
  ],
  "automations": [
    {
      "id": "daily-briefing",
      "agentId": "researcher",
      "templateId": "research-agent/daily-briefing",
      "enabled": true,
      "required": true,
      "schedule": "0 10 * * *",
      "timeZone": "Asia/Shanghai",
      "delivery": {
        "endpointId": "research-feishu",
        "targetRef": "env:DAILY_BRIEFING_TARGET",
        "targetType": "union_id"
      }
    }
  ],
  "backgroundSessions": {
    "enabled": false,
    "required": false,
    "stepTimeoutMs": 300000,
    "maxConcurrentSessions": 4,
    "leaseMs": 30000,
    "leaseRenewalIntervalMs": 10000,
    "maxConsecutiveFailures": 3,
    "retryBackoffMs": 30000,
    "sessionLifetimeMs": 86400000
  },
  "endpoints": [
    {
      "id": "research-feishu",
      "agentId": "researcher",
      "sessionNamespace": "research-feishu-v1",
      "credentialRef": "env:RESEARCH_FEISHU",
      "enabled": true,
      "required": true,
      "baseUrl": "https://open.feishu.cn",
      "streamMinIntervalMs": 200,
      "progressDisplay": "collapsed",
      "groupPolicy": "mention-only"
    }
  ]
}

The application Composition Root supplies Runtime, Endpoint, and Automation adapters:

const daemon = await createConfiguredRivusDeploymentDaemon({
  manifestPath: "./rivus.config.json",
  createRuntime,
  createEndpoint,
  createAutomation
});

await daemon.start(); // rejects when an enabled Required Endpoint or Automation cannot start

Endpoint factories receive a Host-bound handle() function. Inbound payload fields cannot select another Agent or Endpoint. createConfiguredRivusDaemonBootstrap() is the small, explicit single-lane Pi/Feishu composition; manifest deployments use the multi-agent path directly, without a parallel compatibility model.

examples/rivus.config.json and examples/pi-feishu-deployment.bootstrap.ts form a concrete two-bot composition. Each credentialRef: "env:<PREFIX>" resolves only <PREFIX>_APP_ID and <PREFIX>_APP_SECRET; each Endpoint owns its WebSocket client, durable Inbox, bounded concurrent worker, CardKit target/delivery state, Human Interaction JSONL repository, and lifecycle. Its unique Manifest Session Namespace is prepended before the trusted Host route reaches a fair per-session Scheduler, Harness lane, and Pi Session Registry. Startup resolves the Bot Open ID, waits for the SDK onReady handshake, repairs missing terminal CardKit projections, and applies groupPolicy before the durable accept. The example restores a separate event log, Tool operation ledger, and Pi session directory per Agent Instance, plus one deployment-scoped append-only Agent Memory log. Built-in Pi tools, extensions, prompt templates, themes, and ambient Pi context discovery remain disabled. An Agent with a trusted projectSpaceId additionally receives only that Project Space's declared native Skill sources and bounded AGENTS.md View; Pi handles progressive Skill loading and /skill:<name>, while unknown names fail before model execution. Unbound Agents retain the previous inline Rivus Skill reader behavior.

Interactive Endpoints default progressDisplay to collapsed: public assistant narration and inspectable Skill/Tool actions remain in chronological order, and every action owns a small native collapsible panel rather than being collected into one process appendix. expanded opens completed action panels initially and hidden omits them; a running action stays open so its safe parameters and current result are visible. A channel-neutral RunPresentation read model preserves stable steps, status, duration, and safe Tool input/result projections across streaming updates and successor cards. Sensitive fields are redacted, machine paths normalized, and long values explicitly truncated before a channel renderer sees them; private reasoning and raw Tool payloads never enter the presentation IR. Ordinary replies never gain an empty process panel. A rollover freezes the predecessor exactly as last rendered and starts a blank successor that receives only post-breakpoint content. Scheduled Automations use a separate final-only AutomationPresentation IR: Plugins own title, metadata, sections, items, notes, and sources, while the channel renderer owns visual components.

An Endpoint may opt into Feishu's native process message with "experimental": { "cotMessages": true }. The deployment then requires a local RIVUS_FEISHU_COT_BASE_URL; no environment-specific origin is committed or inferred. Rivus creates the process message before the CardKit target so Feishu displays the process above the authoritative final answer, then renders redacted reasoning, model/Skill steps, and Tool calls into AG-UI COT events. Tool arguments remain valid redacted JSON as required by AG-UI, while the Feishu presenter uses narrow mappings for the deployment's known Tools, a sanitized bounded fallback for unknown Tools, and labeled result lines such as 查询天气 · 北京. COT failures are reported and degrade to CardKit instead of failing the Agent Run; if CardKit preparation fails after COT creation, Rivus explicitly closes that process as failed.

The example declares disabled daily-ielts-word, a-share-pre-market, and a-share-post-market Automations. Enable them only in a gitignored local manifest and set each targetRef variable to a Feishu receive ID. The scheduler accepts M H * * * and the weekday-only subset M H * * 1-5 plus an IANA time zone. The A-share templates run at 09:00 and 16:00 on weekdays, request only the bounded evidence-backed briefing Tool, and relay its source-attributed facts and explicitly labeled inferences without letting the model invent market numbers or causal claims; see the A-share briefing runbook. Each occurrence uses an Automation-only Agent Instance and Session, persists generated text before delivery, and sends with a stable Feishu UUID so restart or delivery retry does not create another model answer. An endpoint-less worker Agent may deliver through another enabled Agent's Host-owned Endpoint, allowing one visible Feishu bot to present multiple isolated task Profiles without inheriting their Tool grants or state. It is a local single-writer scheduler, not a distributed cron service.

rivus --env-file .env.local \
  --bootstrap ./examples/pi-feishu-deployment.bootstrap.ts \
  --manifest ./examples/rivus.config.json

The manifest deployment bootstrap enables Langfuse tracing when all three credential variables below are present. It exports one OpenTelemetry trace per Agent Run with nested Turn, model, and Tool spans. Content is metadata-only by default. Set RIVUS_TELEMETRY_CONTENT=redacted to include Run/Turn, authoritative per-model-call, and Tool input/output after recursive redaction and size bounding. Pi snapshots each model request from its effective system prompt, transcript, and active Tool schemas, plus the final assistant message; a transient application port sends those facts directly to the telemetry adapter without adding raw model content to the durable Agent event log. The built-in policy replaces known credential fields, common inline secrets, common personal identifiers, and machine home-directory prefixes. Model thinking, raw deltas, and raw error messages remain excluded. The heuristic remains defense in depth rather than a guarantee for arbitrary domain-specific secrets.

LANGFUSE_BASE_URL=https://jp.cloud.langfuse.com
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
RIVUS_TELEMETRY_ENVIRONMENT=local
RIVUS_TELEMETRY_CONTENT=redacted

Keep these values in .env.local. A partial Langfuse configuration fails at startup so an expected telemetry integration cannot silently drop traces. Export uses the regional Langfuse OTLP/HTTP endpoint and is batch-processed independently from the durable Agent event log; exporter failure does not fail an Agent Run. Daemon shutdown flushes and closes the exporter. createTelemetryContentRedactor() supplies the default bounded policy, and a custom redactContent function can be passed to createOpenTelemetryAgentEventSink() when a deployment needs stricter organization-specific rules.

The shipped example also includes an Open-Meteo current-weather Tool. This one-shot command uses the deployment's default Agent without starting Feishu WebSockets, marks the local operator invocation explicitly, and produces a searchable Agent 执行 -> Agent 处理 -> 模型调用 -> current-weather -> 模型调用 trace. Display names stay readable while standard gen_ai.* attributes preserve portable model, Tool, token, and cost semantics. Local CLI invocations have no human approval actor, so approval-required Tools remain unavailable.

RIVUS_WEATHER_DEFAULT_LOCATION=上海 rivus \
  --env-file .env.local \
  --bootstrap ./examples/pi-feishu-deployment.bootstrap.ts \
  --manifest ./examples/rivus.config.json \
  --prompt "今天天气怎么样" \
  --session-key local:agent-a:weather-demo

The same local prompt path can prove durable Agent Memory without starting Feishu. The Profile and Deployment must both grant agent-private, and the Deployment must set memory.tool: true. The CLI binds the Tool to RIVUS_MEMORY_TENANT_ID, RIVUS_LOCAL_SUBJECT_ID, and the resolved Agent; an optional RIVUS_LOCAL_CONVERSATION_ID supplies a separate trusted Conversation identity when that scope is granted. None of those identities are model arguments. Run two independent processes against the same state directory and use different Session Keys to prove cross-session Agent-private recall:

RIVUS_DEPLOYMENT_STATE_DIR=.rivus/memory-demo \
RIVUS_MEMORY_TENANT_ID=local-demo \
RIVUS_LOCAL_SUBJECT_ID=operator \
rivus --env-file .env.local \
  --bootstrap ./examples/pi-feishu-deployment.bootstrap.ts \
  --manifest ./examples/rivus.config.json \
  --prompt "请调用 rivus_memory 的 propose 命令记住：验收代号是 cedar-47。只报告工具结果。" \
  --session-key local:agent-a:memory-write

RIVUS_DEPLOYMENT_STATE_DIR=.rivus/memory-demo \
RIVUS_MEMORY_TENANT_ID=local-demo \
RIVUS_LOCAL_SUBJECT_ID=operator \
rivus --env-file .env.local \
  --bootstrap ./examples/pi-feishu-deployment.bootstrap.ts \
  --manifest ./examples/rivus.config.json \
  --prompt "请调用 rivus_memory 的 search 命令搜索 cedar-47，并原样告诉我找到的内容。" \
  --session-key local:agent-a:memory-read

propose creates a proposed record; the model cannot call confirm, mark it conversation-safe, write arbitrary Shared User Profile text, or select another tenant, Agent, subject, conversation, audience, or storage path. forget_request may tombstone only an exact still-proposed private or Conversation record; confirmed and shared records require trusted control. Shared Profile search/read hides unconfirmed proposals. A trusted group audience removes Agent-private and Shared Profile scopes before Tool authority is issued, leaving only the current Conversation and confirmed Project scopes when granted. Replayed search/read operations recheck current visibility, so a tombstone cannot leak stale content from the Tool operation fence. The example stores Memory at <RIVUS_DEPLOYMENT_STATE_DIR>/memory/agent-memory.jsonl with owner-only file permissions and rejects a broken revision chain on restore.

The separate examples/rivus-langfuse-demo.config.json grants one dedicated profile a versioned HTML-publishing Skill and exactly two mutation Tools: write one operation-bound artifact and upload only that artifact to the current user's Feishu Drive root through lark-cli. The Runtime supplies one built-in rivus_read_skill reader only when the resolved Skill grant is non-empty; it is not a Plugin Tool and cannot read outside the immutable skillGrantSet. The normal two-bot manifest does not grant these Drive capabilities. Automation templates must request Skills explicitly and receive the exact intersection of Profile, Deployment, and Template grants. The uploader rejects arbitrary paths and symbolic links, then copies a no-follow-opened artifact into a private staging file before invoking lark-cli. Authenticate lark-cli as a user with Drive upload scope before running the demo:

lark-cli auth status --json --verify

rivus --env-file .env.local \
  --bootstrap ./examples/pi-feishu-deployment.bootstrap.ts \
  --manifest ./examples/rivus-langfuse-demo.config.json \
  --prompt "Read the granted Skill, create a polished Langfuse introduction, and upload it as langfuse-guide.html" \
  --session-key local:langfuse-demo:html-publish

This produces a multi-step trace containing 技能 · Langfuse HTML publisher, write-html-artifact, and lark-drive-upload-html between model generations. Durable Skill events record ID, version, digest, title, content length, and outcome without storing the Skill body. The Skill requires synthetic examples to be labeled and execution claims to come from supplied evidence or Tool results. The durable Tool operation ledger replays a completed upload result and fences an uncertain side effect instead of silently uploading twice. The publication boundary is recorded in RFC 0009; the first-class Skill Runtime and E2E evidence contract are recorded in RFC 0010.

Run the opt-in cloud E2E Harness after lark-cli user authentication and .env.local Langfuse configuration. It builds the package, uses a unique Session Key, file name, and content marker, runs the real Agent, inspects and downloads the returned Drive file, then polls Langfuse until the strict model/Skill/model/write/model/upload/model sequence is visible and the successful upload output identifies that same file. Every request is deadline-bounded, and only traces discovered for the unique Session are fetched. The Harness prints only sanitized evidence; it is intentionally not part of credential-free CI.

npm run e2e:langfuse-drive

The package also builds a rivus bin. The bin intentionally loads a bootstrap module instead of importing specific Pi or Feishu SDK packages in the core:

FEISHU_APP_ID=cli_x \
FEISHU_APP_SECRET=secret_x \
rivus --bootstrap ./rivus.bootstrap.mjs

The CLI can also own the manifest-driven startup chain. In this mode the Bootstrap exports createRivusDeploymentAdapters(context) with application-edge createRuntime, createEndpoint, and optional createAutomation factories; the CLI loads the Manifest, runs the Plugin Loader, constructs the Multi-agent Host, and then activates Endpoints followed by Automations:

rivus --bootstrap ./rivus.bootstrap.mjs --manifest ./rivus.config.json

--status reports the complete Deployment without activating Endpoints. --prompt runs the default Agent on the instance bound to defaultEndpointId without starting unrelated Endpoint transports. It supplies an explicit local-cli invocation origin, binds optional Memory identity from RIVUS_MEMORY_TENANT_ID, RIVUS_LOCAL_SUBJECT_ID, and RIVUS_LOCAL_CONVERSATION_ID, and allows granted non-approval Tools to run, while the empty actor set keeps approval-required Tools fail-closed. A Bootstrap may additionally expose a manifest replay adapter for the existing Feishu replay commands.

rivus --manifest ./rivus.config.json --check-config performs side-effect-free schema and semantic validation, including unique identities, references, Session Namespaces, Automation delivery bindings, and the enabled default binding. It prints only the target reference name, never the resolved receive ID. It does not load the Bootstrap or Plugin code and does not require legacy FEISHU_* values.

Manifest deployments may also expose an offline Recovery Control. Stop the daemon before using these commands: the bundled JSON/JSONL stores remain single-writer and do not provide cross-process locking. --recovery-list prints payload-free Dead Letter summaries and Tool operations currently fenced as reconciliation-required; raw worker and provider errors are replaced with bounded failure codes and generic operator guidance. Every mutation requires the exact listed revision plus an audit note, so a repeated or stale operator action fails closed.

rivus --bootstrap ./rivus.bootstrap.mjs --manifest ./rivus.config.json --recovery-list

install -m 600 /dev/null .rivus/recovery-note.txt
${EDITOR:-vi} .rivus/recovery-note.txt

rivus --bootstrap ./rivus.bootstrap.mjs --manifest ./rivus.config.json \
  --requeue-dead-letter om_xxx \
  --endpoint-id feishu-agent-a \
  --expected-revision 5 \
  --recovery-note-file .rivus/recovery-note.txt

Requeue resets the delivery's bounded attempt counter and retains the original payload behind the Inbox boundary. It is rejected while any pending or uncertain Tool operation remains bound to the same source message; reconcile those Tool fences first. An uncertain Tool operation must be reconciled from external evidence. Confirming applied requires the exact stable-JSON result that future retries should replay; confirming not-applied releases the existing fence so the same stable Operation ID may execute again.

install -m 600 /dev/null .rivus/tool-result.json
${EDITOR:-vi} .rivus/tool-result.json

rivus --bootstrap ./rivus.bootstrap.mjs --manifest ./rivus.config.json \
  --resolve-tool-operation operation_xxx \
  --instance-id instance_xxx \
  --expected-revision 2 \
  --tool-outcome applied \
  --tool-result-file .rivus/tool-result.json \
  --recovery-note-file .rivus/recovery-note.txt

Recovery notes and Tool results are intentionally read from regular files with owner-only permissions (0600 or stricter), not command arguments that can leak through shell history or process listings. Mutation output confirms identities, revision, actor, timestamp, and terminal state without echoing the note or applied result.

If you already have an OpenClaw local config, you can print a Rivus-compatible env file instead of copying Feishu and GLM settings by hand:

rivus --print-openclaw-env ./openclaw.json --pi-api-key-file /path/to/local-key.txt > .env.local

The generated file includes Feishu credentials, the selected PI_MODEL, matching provider PI_BASE_URL, explicit OpenClaw thinking level as PI_THINKING_LEVEL when present, and PI_API_KEY_FILE when provided. Treat the output as secret-bearing local config and keep it in a gitignored file.

Then point the CLI at that local file instead of exporting each variable by hand:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs

To validate the local file and BYOK key path without loading a bootstrap module or printing secrets:

rivus --env-file .env.local --check-config

To print the bootstrap status as JSON without starting the long-connection daemon:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --status

To run one local prompt through the bootstrap harness without starting the long-connection daemon:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --prompt "Say hello" --session-key local:main:cli

To replay one Feishu receive-message payload through the bootstrap runtime without starting the long-connection daemon:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --replay-feishu-event ./payload.json

The replay JSON includes an intake section with messageId, commandType, sessionKey, and the tenant/chat/thread/agent reference that produced that key, so a captured Feishu payload can be checked for session mapping before live callbacks are debugged.

To smoke test the same Feishu receive path without a captured payload file, synthesize a text message locally:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --replay-feishu-text "local smoke" --feishu-message-id om_local_smoke

The synthetic replay defaults to oc_cli, omt_cli, and tenant_cli, and accepts --feishu-chat-id, --feishu-thread-id, and --feishu-tenant-key when you need to exercise a specific Feishu session mapping. It disables Feishu run side effects for this one replay, so a fake message id does not create or reply with a CardKit card.

The packaged example also supports live local status while the daemon is running:

RIVUS_STATUS_PORT=17890 \
rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs

From another terminal:

rivus --status-url http://127.0.0.1:17890/status

For a single-command inbound smoke check, start the daemon and wait until the Feishu callback or worker path is observed:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --wait-receive handled --wait-timeout-ms 30000

The bootstrap module should export createRivusDaemonProcess(context) or a default factory and return a started-ready daemon process. If the returned object also exposes promptText(command), rivus --prompt uses that method for local model smoke tests without calling start(). If it exposes replayReceiveMessage(payload), rivus --replay-feishu-event <json> reads the payload file, rivus --replay-feishu-text <text> builds a synthetic text payload, and both commands run the local receive-message replay path, print the replay result as JSON, and exit without calling start(). A typical module creates SDK clients at the edge, then delegates the local-daemon wiring and event-log hydration to restoreConfiguredRivusDaemonBootstrap:

import { Effect } from "effect";
import {
  createJsonFileFeishuCardTargetRegistry,
  createJsonFetchRequest,
  createJsonlAgentEventLog,
  restoreConfiguredRivusDaemonBootstrap
} from "@rivus/agent";

export async function createRivusDaemonProcess(context) {
  const bootstrap = await Effect.runPromise(
    restoreConfiguredRivusDaemonBootstrap({
      cardTargets: createJsonFileFeishuCardTargetRegistry({
        filePath: "./.rivus/feishu-card-targets.json"
      }),
      clock,
      config: context.config,
      eventDispatcher,
      eventLog: createJsonlAgentEventLog({
        filePath: "./.rivus/agent-events.jsonl"
      }),
      loop: piLoop,
      request: createJsonFetchRequest(),
      runIds,
      sleep,
      websocketClient
    })
  );

  return {
    running: () => bootstrap.process.running(),
    start: () => bootstrap.process.start(),
    status: () => bootstrap.status(),
    stop: () => bootstrap.process.stop()
  };
}

Configuration

loadRivusDaemonConfig(env) parses a plain environment object and returns a typed config without reading process.env directly.

Recognized variables:

• FEISHU_APP_ID and FEISHU_APP_SECRET are required.
• FEISHU_BASE_URL defaults to https://open.feishu.cn.
• FEISHU_STREAM_MIN_INTERVAL_MS defaults to 200, matching a conservative 5 QPS stream update envelope.
• RIVUS_AGENT_ID defaults to main.
• PI_API_KEY or PI_API_KEY_FILE, plus PI_BASE_URL, PI_MODEL, and PI_THINKING_LEVEL, are optional Pi adapter settings.
• PI_API_KEY_FILE reads and trims a local key file; it is mutually exclusive with PI_API_KEY.
• PI_THINKING_LEVEL accepts off, minimal, low, medium, high, or xhigh.
• LANGFUSE_BASE_URL, LANGFUSE_PUBLIC_KEY, and LANGFUSE_SECRET_KEY optionally enable metadata-only OpenTelemetry export in the manifest deployment bootstrap and must be set together.
• RIVUS_TELEMETRY_CONTENT defaults to metadata-only; set it explicitly to redacted to export recursively redacted and size-bounded Run/Turn/Model/Tool content. Other values are rejected.
• RIVUS_WEATHER_DEFAULT_LOCATION selects the example current-weather Tool's fallback city and defaults to 北京; it is not a Core setting.
• RIVUS_TELEMETRY_ENVIRONMENT defaults to NODE_ENV, then local; RIVUS_TELEMETRY_SERVICE_NAME defaults to rivus-agent.

For both SDK bootstrap templates, PI_MODEL should use Pi's provider/model form when PI_API_KEY, PI_API_KEY_FILE, or PI_BASE_URL is set, for example anthropic/claude-opus-4-5 or zai/glm-5.1 for GLM BYOK. The templates apply the resolved Pi API key through Pi AuthStorage.setRuntimeApiKey(), resolve the model through Pi's public ModelRegistry.find(provider, modelId) API, and apply PI_BASE_URL by writing a local .rivus/pi-models.json provider override, matching Pi's models.json configuration model.

Core Usage

External Agent Plugin definitions

External Agent packages can register declarative Profiles, Tools, Skills, and Automation templates without starting runtime resources. A Host-owned Deployment then narrows the Profile's requested Tools with an exact allowlist:

import {
  createRivusPluginCatalog,
  resolveRivusAgentDefinition,
  RIVUS_PLUGIN_API_VERSION
} from "@rivus/agent";

const catalog = createRivusPluginCatalog();
catalog.registerPlugin({
  manifest: {
    apiVersion: RIVUS_PLUGIN_API_VERSION,
    id: "example-agent",
    version: "1.0.0"
  },
  register(registry) {
    registry.registerTool({
      createExecutor: () => ({ execute: async () => ({ ok: true }) }),
      description: "Read an example source",
      digest: "sha256:example-read-v1",
      id: "example-agent/read",
      idempotency: "supported",
      inputSchema: { type: "object" },
      risk: "observe",
      version: "1.0.0"
    });
    registry.registerAgentProfile({
      displayName: "Example Agent",
      id: "example",
      memory: { scopes: [] },
      model: { provider: "example" },
      skills: { allow: [] },
      systemPrompt: "Use trusted example sources.",
      tools: { allow: ["example-agent/read"] }
    });
  }
});

const definition = resolveRivusAgentDefinition(catalog, {
  agentId: "example",
  endpointIds: ["example-feishu"],
  pluginId: "example-agent",
  profileId: "example",
  tools: { allow: ["example-agent/read"] }
});

The resolved definition exposes immutable Tool schemas and a ToolGrantSet; executor factories remain inside the Catalog boundary. Wildcards, duplicate identifiers, unknown references, and incompatible Plugin API versions are rejected.

External Plugin packages can run reusable conformance checks from the testing subpath:

import { assertRivusPluginConforms } from "@rivus/agent/testing";

await assertRivusPluginConforms({ plugin, deployment });

Agent Harness

import { Effect, Stream } from "effect";
import {
  createAgentHarnessClient,
  createAgentRuntime,
  createAgentDomainEventHandler,
  createAgentDomainEventSinkFromCallback,
  createAgentRunUpdateHandler,
  createDefaultAgentHarnessClientFromTextCallback,
  createDefaultAgentHarnessFromTextCallback,
  createDefaultAgentRuntimeFromTextCallback,
  createJsonlAgentEventLog,
  isAssistantTextDeltaEvent,
  restoreAgentHistory
} from "@rivus/agent";

const eventLog = createJsonlAgentEventLog({
  filePath: "./.rivus/agent-events.jsonl"
});

const auditSink = createAgentDomainEventSinkFromCallback((event) => {
  console.log(event.type);
});

const harness = createDefaultAgentHarnessFromTextCallback(
  (input) => `Hello ${input.text}`,
  {
    eventSinks: [eventLog, auditSink]
  }
);

harness.subscribe((event) => {
  if (isAssistantTextDeltaEvent(event)) {
    console.log(event.delta);
  }
  console.log(event.type);
});

const finalText = await Effect.runPromise(
  harness.promptText({
    sessionKey: "feishu:tenant:chat:thread",
    text: "Say hello"
  })
);

console.log(finalText);

const runtime = createAgentRuntime(harness, {
  mainSessionKey: "feishu:tenant:chat:thread"
});
console.log(await runtime.promptText("Say hello from the main session"));
console.log((await runtime.promptTurn("Return a turn")).assistantText);

const chat = harness.forSession("feishu:tenant:chat:thread");
const unsubscribeChat = chat.subscribe((event) => {
  console.log(event.type);
});
const unsubscribeChatUpdates = chat.subscribeUpdates(({ state }) => {
  console.log(state.phase, state.finalText);
});
const unsubscribeRunUpdates = chat.subscribeRunUpdates("run_123", ({ state }) => {
  console.log(state.phase);
});
const chatText = await Effect.runPromise(chat.promptText("Say hello"));
const chatTurn = await Effect.runPromise(chat.promptTurn("Say hello"));
const chatPromptSnapshot = await Effect.runPromise(chat.promptSnapshot("Say hello"));
const chatRunSnapshot = await Effect.runPromise(chat.promptRunSnapshot("Say hello"));
const chatResult = await Effect.runPromise(chat.prompt("Say hello"));
const chatSnapshot = chat.getSnapshot();
console.log(chatSnapshot.availability.busy, chatSnapshot.state?.finalText);
console.log(chatTurn.assistantText);
console.log(chatPromptSnapshot.summary?.finalText);
console.log(chatRunSnapshot.summary.finalText);
console.log(chatResult.snapshot.state?.finalText);
console.log(chatResult.runSnapshot.summary.finalText);
console.log(chat.getRunSnapshot(chatResult.runId)?.summary.finalText);
await Effect.runPromise(
  chat.promptWithEvents(
    "Say hello",
    createAgentDomainEventHandler((event) => {
      console.log(event.type);
    })
  )
);
await Effect.runPromise(
  chat.promptWithUpdates(
    "Say hello",
    createAgentRunUpdateHandler(async ({ state }) => {
      await Promise.resolve();
      console.log(state.phase, state.finalText);
    })
  )
);
await Effect.runPromise(
  Stream.runForEach(chat.stream("Say hello"), (event) =>
    Effect.sync(() => {
      console.log(event.type);
    })
  )
);
await Effect.runPromise(
  Stream.runForEach(chat.streamText("Say hello"), (delta) =>
    Effect.sync(() => {
      console.log(delta);
    })
  )
);
await Effect.runPromise(
  Stream.runForEach(chat.streamUpdates("Say hello"), ({ state }) =>
    Effect.sync(() => {
      console.log(state.phase, state.finalText);
    })
  )
);
await Effect.runPromise(chat.cancelRun("run_from_card", "user cancelled"));

console.log(chat.sessionKey);
console.log(chatText);
console.log(chatResult.updates.at(-1)?.state.phase);
console.log(chat.getAvailability().busy);
console.log(chat.getActiveRun()?.runId);
console.log(chat.getActiveRunState()?.phase);
console.log(chat.getHistory().totalEvents);
console.log(chat.getSummary()?.runCount);
console.log(chat.getRuns().map((run) => run.runId));
console.log(chat.getLatestRunSummary()?.runId);
console.log(chat.getLatestRunSnapshot()?.summary.finalText);
console.log(chat.getRunSummary(chatResult.runId)?.prompt);
console.log(chat.getRunEvents(chatResult.runId)?.length);
console.log(chat.getRunUpdates(chatResult.runId)?.at(-1)?.state.phase);
console.log(chat.getRunState(chatResult.runId)?.phase);
console.log(chat.getState()?.phase);

const client = createAgentHarnessClient(harness);
const chatClient = client.forSession("feishu:tenant:chat:thread");
const promiseText = await chatClient.promptText("Say hello from a Promise adapter");
const promiseTurn = await chatClient.promptTurn("Say hello from a Promise adapter");
const promiseAttempt = await chatClient.tryPromptText("Try without throwing");
if (promiseAttempt.ok) {
  console.log(promiseAttempt.value);
} else {
  console.log(promiseAttempt.error);
}
await chatClient.promptWithUpdates("Stream progress", async ({ state }) => {
  console.log(state.phase, state.finalText);
});
await chatClient.promptWithText("Stream text with final result", async (delta) => {
  console.log(delta);
});
for await (const delta of chatClient.streamText("Stream text")) {
  console.log(delta);
}
console.log(promiseText);
console.log(promiseTurn.assistantText);

const quickClient = createDefaultAgentHarnessClientFromTextCallback((input) => `Hello ${input.text}`);
console.log(await quickClient.forSession("feishu:tenant:chat:thread").promptText("Say hello"));

const quickRuntime = createDefaultAgentRuntimeFromTextCallback((input) => `Hello ${input.text}`, undefined, {
  mainSessionKey: "feishu:tenant:chat:thread"
});
console.log(await quickRuntime.promptText("Say hello from the runtime"));
console.log(await quickRuntime.mainClient.promptText("Say hello from a Promise adapter"));
console.log(await Effect.runPromise(quickRuntime.mainSession.promptText("Say hello from Effect")));

unsubscribeChat();
unsubscribeChatUpdates();
unsubscribeRunUpdates();

const result = await Effect.runPromise(
  harness.prompt({
    sessionKey: "feishu:tenant:chat:thread",
    text: "Say hello"
  })
);

console.log(result.finalText);
console.log(result.updates.at(-1)?.state.phase);
console.log(harness.getActiveRun()?.sessionKey);
console.log(harness.getActiveRunId());
console.log(harness.getHistory().totalEvents);
console.log(harness.getRunEvents(result.runId)?.map((event) => event.type));
console.log(harness.getRunUpdates(result.runId)?.map((update) => update.state.phase));
console.log(harness.getRunSummary(result.runId)?.toolExecutionCount);
console.log(harness.getRunState(result.runId)?.phase);
console.log(harness.getSessionHistory("feishu:tenant:chat:thread").totalEvents);
console.log(harness.getSessionSummary("feishu:tenant:chat:thread")?.latestRunId);
console.log(harness.getSessionRuns("feishu:tenant:chat:thread").map((run) => run.runId));
console.log(harness.getSessionLatestRunSummary("feishu:tenant:chat:thread")?.runId);
console.log(harness.getSessionLatestRunSnapshot("feishu:tenant:chat:thread")?.summary.finalText);
console.log(harness.getSessionState("feishu:tenant:chat:thread")?.finalText);

const streamedResult = await Effect.runPromise(
  harness.promptWithEvents(
    {
      sessionKey: "feishu:tenant:chat:thread",
      text: "Say hello"
    },
    createAgentDomainEventHandler((event) => {
      console.log(event.type);
      console.log(harness.getActiveRunState()?.finalText);
    })
  )
);

console.log(streamedResult.finalText);

await Effect.runPromise(
  harness.promptWithUpdates(
    {
      sessionKey: "feishu:tenant:chat:thread",
      text: "Say hello"
    },
    createAgentRunUpdateHandler(({ event, state }) => {
      console.log(event.type, state.phase, state.finalText);
    })
  )
);

await Effect.runPromise(
  Stream.runForEach(
    harness.stream({
      sessionKey: "feishu:tenant:chat:thread",
      text: "Say hello"
    }),
    (event) =>
      Effect.sync(() => {
        console.log(event.type);
      })
  )
);

await Effect.runPromise(
  Stream.runForEach(
    harness.streamText({
      sessionKey: "feishu:tenant:chat:thread",
      text: "Say hello"
    }),
    (delta) =>
      Effect.sync(() => {
        console.log(delta);
      })
  )
);

await Effect.runPromise(
  Stream.runForEach(
    harness.streamUpdates({
      sessionKey: "feishu:tenant:chat:thread",
      text: "Say hello"
    }),
    ({ event, state }) =>
      Effect.sync(() => {
        console.log(event.type, state.phase, state.finalText);
      })
  )
);

createAgentHarnessClient(harness) is an optional Promise facade for non-Effect adapters. The core harness remains Effect-first, while the client exposes Promise-returning prompt, callback, async-iterable stream, and cancellation helpers plus the same synchronous read-model and subscription methods. promptWithText() filters ordered domain events to assistant text deltas while still returning the final prompt result, which keeps IM progress renderers out of event-type plumbing. tryPrompt() and tryPromptText() return { ok: true, value } or { ok: false, error }, so non-Effect adapters can branch on harness failures without exception control flow. createDefaultAgentHarnessClient(), createDefaultAgentHarnessClientFromCallback(), and createDefaultAgentHarnessClientFromTextCallback() compose the same facade with default clock/run-id wiring for one-step local bootstraps. createAgentRuntime(harness) wraps any existing harness in a tiny session-first facade: call runtime.promptText(), runtime.promptTurn(), or runtime.tryPromptText() for the main session directly, use mainClient when an application wants the full Promise session client, use mainSession when it wants Effect helpers, and use harness when it needs the full lower-level use case. createDefaultAgentRuntime(), createDefaultAgentRuntimeFromCallback(), and createDefaultAgentRuntimeFromTextCallback() compose that same facade with default harness wiring.

Use promptTurn() when an adapter wants the completed user/assistant turn directly. It follows the same prompt path as promptText() and prompt(), and returns the replayed AgentTranscriptTurn with user text, assistant text, thinking text, phase, timing, and tool counts.

createDefaultAgentHarness wires a system clock and UUID run ids by default; pass clock or runIds when deterministic tests or custom runtime boundaries need them. createDefaultAgentHarnessFromCallback(run, options) adds the same default runtime wiring while accepting a callback-shaped event source directly; createDefaultAgentHarnessFromTextCallback(generate, options) is the shortest bootstrap when a local SDK or model call only returns final assistant text. Pass initialEvents when a process has restored append-only domain events and wants the new harness to seed run-event timelines, process-local history snapshots, transcript snapshots, and same-session continuation before accepting work; pass initialRunStates when only projected run/session indexes are available. createFeishuAgentRuntime forwards both options to its internal harness. Restored non-terminal states remain available through getRunState(runId) but are not used as same-session continuation state, so a crash-time partial run is not treated as the previous answer. AgentLoopInput.previousSessionState gives loop adapters the latest terminal projected AgentRunState for the same sessionKey before the new run is accepted, AgentLoopInput.previousSessionTranscript gives them the already completed same-session turns before the current prompt, and AgentLoopInput.messages gives stateless SDK/RPC adapters provider-friendly { role, content } conversation messages derived from that transcript plus the current user prompt; adapters do not need to query the harness for those views. AgentLoopInput.abortSignal is aborted when a harness cancellation request succeeds, so SDK and RPC adapters can stop underlying work in addition to letting the Effect stream unwind. forSession(sessionKey) returns a small AgentSessionHandle that binds promptText(text), promptSnapshot(text), promptRunSnapshot(text), prompt(text), promptWithEvents(text, onEvent), promptWithUpdates(text, onUpdate), stream(text), streamUpdates(text), cancelActiveRun(reason), cancelRun(runId, reason), best-effort event/update subscription, run-specific event/update subscription, availability lookup, active-run lookup, active-state lookup, history lookup, transcript lookup, session-summary lookup, run-list lookup, latest-run-summary lookup, latest-run-snapshot lookup, run-summary lookup, run-snapshot lookup, run-event lookup, run-update lookup, run-state lookup, session-state lookup, and a combined session snapshot to one stable session key, which is useful for IM adapters that should not repeatedly rebuild the same key. promptText() is the shortest path for callers that only need final assistant text while still using the same domain events, sinks, busy gate, and error model as prompt(). promptSnapshot() runs the same prompt path but returns only the completed AgentSessionSnapshot, which is useful when an adapter wants the final session read model instead of event/update history. promptRunSnapshot() runs the same prompt path but returns only the completed AgentRunSnapshot, which is useful when an adapter wants one run's detail read model instead of the session aggregate. prompt() returns accepted domain events, ordered AgentRunUpdate values, the completed session snapshot, and the completed run snapshot, so result-only callers can inspect final read models and event/state projection timeline without registering a handler or re-querying the harness. promptWithEvents() is the adapter-friendly path when a caller needs ordered domain-event side effects and the final AgentPromptResult from the same run. promptWithUpdates() carries each ordered domain event together with the prior AgentRunState when one exists and the state after that event has been projected, so adapters that render progress do not need to re-read harness state or keep local previous-frame bookkeeping. createAgentDomainEventHandler() and createAgentRunUpdateHandler() adapt sync or async callbacks into those ordered prompt handlers without changing the harness failure model. stream() exposes the ordered domain-event sequence as an Effect stream, and streamUpdates() exposes the same event-plus-state sequence for adapters that prefer stream composition over callbacks; interrupting either stream aborts the loop input signal, records an internal agent_run_cancelled event for sinks, subscribers, and read models, and releases the active run without emitting that terminal event to the interrupted consumer. assistant_thinking_delta events are projected into AgentRunState.thinkingText separately from finalText; Feishu stream projection still publishes only assistant text by default. eventSinks are awaited before the harness updates in-memory state or notifies subscribers; use createAgentDomainEventSink() for Effect-backed sinks or createAgentDomainEventSinkFromCallback() for sync or async callback sinks when a reliable side effect should share that sink contract. If a sink fails, including a synchronous append() throw before an Effect is returned, the run fails with AgentEventSinkFailed and no partially exposed event is published. Loop failures, including synchronous AgentLoop.run() throws before a stream is returned, are reported as AgentLoopFailed with the failed AgentRunState, including any partial text already emitted. Ordered event handler failures are reported as AgentEventHandlerFailed with the current AgentRunState after the triggering event has been projected. Busy rejections are reported as AgentHarnessBusy with the active { runId, sessionKey } plus activeState when a projection is already available, so adapters can render the current phase and partial text directly. cancelActiveRun(reason) requests cancellation for the current busy run, while cancelRun(runId, reason) cancels only when the supplied run id still matches the active run; forSession(sessionKey).cancelActiveRun(reason) additionally requires the active run to belong to that session, and forSession(sessionKey).cancelRun(runId, reason) requires both the session and exact run id to match. Successful cancellation paths record agent_run_cancelled, project the run to cancelled, and fail the caller with AgentRunCancelled carrying partial state. Use exact run-id cancellation for stale-prone external UI or card callbacks that may arrive after a newer run starts; use session-active cancellation only for trusted same-session controls that intentionally cancel the currently visible run. subscribe() and subscribeUpdates() callbacks are best-effort observers; thrown listener errors do not fail runs, so reliable side effects should use eventSinks. subscribeRun(runId) and subscribeRunUpdates(runId) apply the same observer semantics while forwarding only one run; forSession(sessionKey).subscribe(), subscribeUpdates(), subscribeRun(runId), and subscribeRunUpdates(runId) additionally require the projected run state to belong to the bound session, including text/tool/terminal events that do not carry sessionKey themselves. getAvailability() exposes a single idle/busy snapshot for adapters that want to render or reject work without probing by starting another prompt. forSession(sessionKey).getAvailability() returns the same idle state when the harness is idle, includes active details only when the active run belongs to that session, and otherwise returns busy: true with activeInSession: false. getActiveRun() exposes the current busy { runId, sessionKey }, getActiveRunId() remains as a convenience for run-id-only callers, and getActiveRunState() exposes the in-flight AgentRunState only while a run is active. forSession(sessionKey).getActiveRun() and forSession(sessionKey).getActiveRunState() return those active projections only when the active run belongs to that session. getHistory(), getSessionHistory(sessionKey), getSessionTranscript(sessionKey), getSessionSnapshot(sessionKey), getSessionSummary(sessionKey), getSessionRuns(sessionKey), getSessionLatestRunSummary(sessionKey), getSessionLatestRunSnapshot(sessionKey), getRunSnapshot(runId), getRunSummary(runId), getRunEvents(runId), getRunUpdates(runId), and getRunState(runId) expose process-local read-model copies for adapters while the process is alive; getSessionSnapshot(sessionKey) returns one session's availability, latest state, aggregate summary, run list, latest run summary, and transcript in one session-safe object; getSessionTranscript(sessionKey) returns user/assistant turns replayed from domain events for one external conversation, createAgentTranscriptMessages(transcript) formats completed turns as { role, content } messages, createAgentConversationMessages({ transcript, text }) appends the current user prompt to those messages, getSessionSummary(sessionKey) returns the aggregate run count, latest run id, phase, text, thinking, and tool counts for one external conversation, getSessionRuns(sessionKey) returns just the session-scoped run summaries for list/detail UIs, getSessionLatestRunSummary(sessionKey) returns the latest run summary for status surfaces, getSessionLatestRunSnapshot(sessionKey) returns the latest run detail snapshot for session-scoped detail surfaces, getRunSnapshot(runId) returns one run's events, ordered update timeline, latest state, and summary for detail surfaces, getRunSummary(runId) returns one run's prompt, timing, text, thinking, and tool-count summary, and getRunUpdates(runId) replays the same event/state update timeline shape returned by prompt(). getSessionState(sessionKey) exposes the latest session projection, and forSession(sessionKey).getHistory(), getTranscript(), getSnapshot(), getSummary(), getRuns(), getLatestRunSummary(), getLatestRunSnapshot(), getRunSnapshot(runId), getRunSummary(runId), getRunEvents(runId), getRunUpdates(runId), and getRunState(runId) return data only for the bound session. eventLog.readRun(runId) returns persisted events that can be replayed with evolveAgentRun.

streamText(command) and session-bound streamText(text) expose only ordered assistant text deltas while reusing the same prompt path, sink ordering, busy gate, cancellation behavior, and state projection as stream(). assistant_thinking_delta still updates AgentRunState.thinkingText, but text-only stream consumers receive only assistant text.

Adapters that can expose model/runtime events through a callback can stay thin with createAgentLoopFromCallback:

import { createAgentLoopFromCallback } from "@rivus/agent";

const loop = createAgentLoopFromCallback(async function* (input) {
  for await (const event of externalRuntime.run(input.text, { signal: input.abortSignal })) {
    yield event;
  }
});

When an adapter only has assistant text chunks, it can return or yield strings; the loop normalizes them to assistant_text_delta events before the harness sees them.

Effect-native adapters can return an Effect from the same helper:

import { Effect } from "effect";
import { createAgentLoopFromCallback } from "@rivus/agent";

const loop = createAgentLoopFromCallback((input) =>
  Effect.succeed([`Hello ${input.text}`])
);

They can also return an Effect Stream directly:

import { Stream } from "effect";
import { createAgentLoopFromCallback } from "@rivus/agent";

const loop = createAgentLoopFromCallback((input) =>
  Stream.fromIterable([`Hello ${input.text}`])
);

Adapters that already receive an async sequence of model/runtime events can also use createAsyncIterableAgentLoop:

import { createAsyncIterableAgentLoop } from "@rivus/agent";

const loop = createAsyncIterableAgentLoop({
  run: async function* (input) {
    for await (const event of externalRuntime.run(input.text)) {
      yield event;
    }
  }
});

Simple demos or SDK calls that return final text can use createTextAgentLoopFromCallback:

import { createTextAgentLoopFromCallback } from "@rivus/agent";

const loop = createTextAgentLoopFromCallback(async (input) => {
  return externalModel.generate(input.text);
});

Effect-native text generators can use createTextAgentLoop:

import { Effect } from "effect";
import { createTextAgentLoop } from "@rivus/agent";

const loop = createTextAgentLoop({
  generate: (input) => Effect.succeed(`Hello ${input.text}`)
});

Local demos and unit tests with a known sync or async script can use createEventAgentLoop:

import { createAgentLoopThinkingDelta, createEventAgentLoop } from "@rivus/agent";

const loop = createEventAgentLoop({
  events: [
    createAgentLoopThinkingDelta("Thinking..."),
    "Hello"
  ]
});

const inputDrivenLoop = createEventAgentLoop({
  events: (input) => [`Hello ${input.text}`]
});

const asyncInputDrivenLoop = createEventAgentLoop({
  events: async (input) => [`Hello ${input.text}`]
});

At daemon startup, restoreAgentHistory(eventLog) reads all persisted domain events and returns run/session summaries for adapters or CLI status views. Run and session summaries expose finalText, thinkingText, toolExecutionCount, and failedToolExecutionCount, so status surfaces can render common projections without replaying state themselves:

const history = await Effect.runPromise(restoreAgentHistory(eventLog));

console.log(history.sessions[0]?.latestRunId);

restoreConfiguredRivusDaemonBootstrap() reads the same event log into AgentHarness.initialEvents, so bootstrap.runtime.harness.getHistory(), getSessionHistory(sessionKey), getRunEvents(runId), and getRunUpdates(runId) include restored events before the worker accepts new Feishu messages.

The configured bootstrap also exposes status() for daemon-facing health and status surfaces:

const status = await Effect.runPromise(bootstrap.status());

console.log(status.harness.busy, status.harness.activeRunState?.finalText, status.queue.pending);

The CLI --status path uses that same method when the bootstrap returns a daemon with status(). It does not start the WebSocket client or worker loop, so it is safe for local inspection of persisted history and bootstrap-visible state. Live status is derived from AgentHarness.getAvailability() and includes the active run state when a run is currently draining, including partial text and tool execution projection.

The CLI replay paths also avoid starting long-running transports. --replay-feishu-event <json> is for captured Feishu payloads, while --replay-feishu-text <text> creates a local im.message.receive_v1-shaped text payload and calls replayReceiveMessage(payload, { sideEffects: "disabled" }). Unlike --prompt, the synthetic replay still resolves the Feishu sessionKey from tenant/chat/thread/agent fields, which makes it useful for checking OpenClaw-style session mapping before a real callback is available without replying to a fake Feishu message id.

For live status from an already-running daemon, inject createRivusDaemonStatusHttpServer through createStatusTransport on createConfiguredRivusDaemonBootstrap. The example bootstrap wires this when RIVUS_STATUS_PORT is set, and rivus --status-url <url> prints that live status JSON without loading a bootstrap module or requiring Feishu credentials. Configured status includes receive.lastAccepted with the latest raw Feishu message id/chat/thread/tenant seen on the fast-ack path, and receive.lastHandled with the latest worker-handled message plus resolved intake.sessionKey. Both receive observations include observedAt, rendered as an ISO timestamp in JSON, so timeout diagnostics can distinguish fresh callbacks from stale state.

During live Feishu E2E checks, --wait-receive accepted waits until the long-connection callback has reached the fast-ack path, while --wait-receive handled waits until the worker has resolved the session and completed handling. Use it with --status-url when the daemon is already running, or with --bootstrap to start the daemon, wait, print the final status JSON, and stop it automatically:

rivus --status-url http://127.0.0.1:17890/status --wait-receive handled --wait-timeout-ms 30000
rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --wait-receive handled --wait-timeout-ms 30000

To avoid accepting an older handled observation during live E2E, send a Feishu message with a unique marker and wait for that marker in receive.lastHandled.intake.text:

rivus --env-file .env.local --bootstrap ./rivus.bootstrap.mjs --wait-receive handled --wait-receive-text rivus-live-marker-123 --wait-timeout-ms 30000

If the sender API returns a Feishu message_id, wait for that exact callback instead of matching only text:

rivus --status-url http://127.0.0.1:17890/status --wait-receive accepted --wait-receive-message-id om_xxx --wait-receive-observed-after 2026-07-09T00:00:00.000Z --wait-timeout-ms 30000
rivus --status-url http://127.0.0.1:17890/status --wait-receive handled --wait-receive-message-id om_xxx --wait-receive-text rivus-live-marker-123 --wait-receive-observed-after 2026-07-09T00:00:00.000Z --wait-timeout-ms 30000

For the full local sequence from daemon startup through user-message send and exact callback verification, use the Feishu live inbound E2E runbook in English or [简体中文](docs/zh-CN/operations/