Initialize Hermes LineUp adapter

This commit is contained in:
2026-08-07 13:35:27 +08:00
commit b9137e176e
17 changed files with 3788 additions and 0 deletions
+5
View File
@@ -0,0 +1,5 @@
__pycache__/
*.py[cod]
.pytest_cache/
.venv/
.env
+9
View File
@@ -0,0 +1,9 @@
# Private runtime configuration for run-hermes-adapter.sh.
# Copy to .env, use the same shared secret as lineup-app-server/.env, and
# never commit either file.
LINEUP_APPSERVER_URL=http://127.0.0.1:8090
LINEUP_AGENT_UID=agent_hermes_main
LINEUP_AGENT_CHANNEL_ID=agent_default_channel
LINEUP_CHANNEL_TYPE=2
LINEUP_AGENT_SHARED_SECRET=
LINEUP_IGNORED_SENDER_UIDS=agent_default
+78
View File
@@ -0,0 +1,78 @@
# LineUp Hermes Adapter
This is the real Hermes integration. It uses the LineUp app server as its
transport boundary and invokes the
locally installed `hermes` CLI for each isolated LineUp conversation.
## Install and enable
```bash
cd lineup-adapter/hermes/lineup
./scripts/install.sh
hermes plugins enable lineup
```
The installer copies only this plugin directory into
`~/.hermes/plugins/lineup`; it never copies local state, logs, or credentials.
## Configure and start
Copy `.env.example` to a private `.env` file (the runner loads it) or set the
same values from a service configuration:
```bash
export LINEUP_APPSERVER_URL=http://127.0.0.1:8090
export LINEUP_AGENT_UID=agent_hermes_main
export LINEUP_AGENT_CHANNEL_ID=agent_default_channel
export LINEUP_AGENT_SHARED_SECRET='replace-with-a-random-secret'
./run-hermes-adapter.sh start
```
Set the same `LINEUP_AGENT_SHARED_SECRET` in `lineup-app-server/.env`. The
server maps it to `agent.shared_secret` and accepts the Adapter heartbeat only
through `X-LineUp-Agent-Secret`.
The initial configuration uses the existing group channel. Restrict ignored
sender identities only when a separately deployed Adapter requires it.
## Guarantees in this version
- Input is consumed through `POST /messages/sync` on the LineUp app server,
not directly through WuKongIM's management API.
- Output is delivered through `POST /messages/send` on the app server.
- SQLite stores the received-message inbox, delivery outbox, and sequence
cursor. A restarted Adapter retries interrupted turns and unsent output.
- The first normal start advances to the current channel sequence and never
replays retained history into Hermes. Set `LINEUP_REPLAY_HISTORY=true` only
for an intentional controlled replay.
- Hermes sessions are named from a hash of the LineUp conversation and user;
the adapter derives that conversation from its configured channel and the
transport sender, so a client-supplied envelope `conversation_id` cannot
switch a user's `--continue` context.
- A newer message from the same user terminates that user's previous local
Hermes subprocess. Timeout is controlled by `LINEUP_HERMES_TIMEOUT_SECONDS`.
- All new outgoing payloads use the `lineup.v1.*` envelope. The input decoder
retains a narrow compatibility bridge for previously stored legacy messages.
- Standard M1 calls are strictly limited to `lineup.v1.tool.call` with
`choice`, `confirm`, or `input` declarations. Every call has a future
`expires_at`, is canonicalized before delivery, and is recorded by
`(conversation_id, call_id)` in SQLite before it is sent.
- `lineup.v1.tool.result` and `lineup.v1.tool.cancel` are consumed at most
once for that durable key. Duplicate, unknown, malformed, and expired
responses do not start another Hermes turn. The audit ledger records only
call id, conversation, event, disposition, and time; it never stores form
values. Hermes receives only the schema-bound result values it needs, not
client envelope metadata, labels, DOM/UI configuration, or audit data.
- This Adapter does not authorize `ui.*`, `app.*`, Surface, capability, or
bundle messages. Those boundaries remain future Milestones.
## Operations
```bash
./run-hermes-adapter.sh check
./run-hermes-adapter.sh start
./run-hermes-adapter.sh status
./run-hermes-adapter.sh stop
```
The retired JavaScript demo Agent is not part of this repository or runtime.
+41
View File
@@ -0,0 +1,41 @@
"""Hermes plugin entry point for the LineUp adapter.
The transport worker deliberately runs out of process (``core.py``). Hermes
loads this module for every normal session, including the child one-shot
processes launched by the worker. Starting a polling thread unconditionally
here would create a worker per reply, so auto-start is explicit and child
processes set ``LINEUP_ADAPTER_CHILD=1``.
"""
from __future__ import annotations
import os
def _status_command(_raw_args: str = "") -> str:
from .core import AdapterConfig, StateStore
config = AdapterConfig.from_env()
store = StateStore(config.state_path)
try:
return store.status_summary()
finally:
store.close()
def register(ctx) -> None:
"""Register a lightweight in-session status command and optional worker."""
ctx.register_command(
"lineup-status",
_status_command,
description="Show LineUp Adapter delivery and queue state.",
)
if (
os.environ.get("LINEUP_ADAPTER_AUTOSTART", "").lower() in {"1", "true", "yes"}
and os.environ.get("LINEUP_ADAPTER_CHILD") != "1"
):
from .core import start_background
start_background()
+422
View File
@@ -0,0 +1,422 @@
"""Small, dependency-free ACP client for the LineUp Hermes Adapter.
Hermes owns the ACP server. This module is intentionally only its host-side
transport: it preserves real tool lifecycle notifications and never derives
activity from model text, status text, or Hermes' later-written SQLite ledger.
"""
from __future__ import annotations
import json
import logging
import os
import queue
import subprocess
import threading
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping
LOG = logging.getLogger("lineup.adapter.acp")
_ID = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.:-"
class ACPError(RuntimeError):
"""A failed, malformed, or unavailable ACP request."""
@dataclass(frozen=True)
class ACPToolUpdate:
"""A privacy-filtered, display-safe real tool lifecycle event."""
tool_id: str
title: str
status: str # running | completed | failed
@dataclass(frozen=True)
class ACPInteractiveOption:
"""One user-visible option projected from an ACP interactive request."""
action_id: str
option_id: str
label: str
@dataclass(frozen=True)
class ACPInteractiveRequest:
"""A bounded Hermes interactive request ready for Adapter-side projection."""
kind: str
request_id: int
session_id: str
title: str
prompt: str
options: tuple[ACPInteractiveOption, ...] = ()
allow_other: bool = False
expects_text: bool = False
multi_select: bool = False
# Backward-compatible aliases while the rest of the adapter migrates from the
# initial permission-only bridge to the generic interaction bridge.
ACPPermissionOption = ACPInteractiveOption
ACPPermissionRequest = ACPInteractiveRequest
_PUBLIC_LABELS_BY_KIND = {
"read": "读取相关配置",
"search": "检索相关文件",
"fetch": "读取公开网页内容",
"execute": "执行已批准的系统检查",
}
def _is_identifier(value: object) -> bool:
return (
isinstance(value, str)
and bool(value)
and len(value) <= 128
and value[0].isalpha()
and all(char in _ID for char in value)
)
def public_tool_update(
update: object, known_titles: Mapping[str, str] | None = None,
) -> ACPToolUpdate | None:
"""Map a real ACP tool event to a bounded public label.
ACP titles, content, raw output and arguments may contain commands, paths,
URLs, credentials, or model-provided text. None is trusted here. The
only inspected values are the ACP's closed tool-kind enum and a previously
stored fixed product label. Hermes deliberately omits ``kind`` from a
completion event, so its tool-call ID may only look up that earlier fixed
label; unknown tools have no public execution step.
"""
if not isinstance(update, dict):
return None
kind = update.get("sessionUpdate")
if kind not in {"tool_call", "tool_call_update"}:
return None
tool_id = update.get("toolCallId")
if (not isinstance(tool_id, str) or not tool_id or len(tool_id) > 128
or tool_id[0] not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
or any(char not in _ID for char in tool_id)):
return None
if kind == "tool_call":
tool_kind = update.get("kind")
if not isinstance(tool_kind, str):
return None
label = _PUBLIC_LABELS_BY_KIND.get(tool_kind)
return ACPToolUpdate(tool_id, label, "running") if label else None
status = update.get("status")
if status == "completed":
projected_status = "completed"
elif status == "failed":
projected_status = "failed"
else:
return None
label = known_titles.get(tool_id) if known_titles else None
return ACPToolUpdate(tool_id, label, projected_status) if label in _PUBLIC_LABELS_BY_KIND.values() else None
class HermesACPClient:
"""Thread-safe ACP stdio client with one persistent Hermes process.
Requests are synchronous from the Adapter's polling loop, while prompt
execution is started in a separate thread so ``session/update`` tool
notifications can be forwarded immediately. The class has no dependency
on the ACP Python package: the installed ``hermes acp`` command owns that
dependency and stdout remains newline-delimited JSON-RPC.
"""
def __init__(
self,
hermes_bin: str,
cwd: str,
on_update: Callable[[dict[str, Any]], None],
on_request: Callable[[ACPInteractiveRequest], None],
on_closed: Callable[[Exception | None], None],
) -> None:
self.hermes_bin = hermes_bin
self.cwd = str(Path(cwd).resolve())
self.on_update = on_update
self.on_request = on_request
self.on_closed = on_closed
self.process: subprocess.Popen[str] | None = None
self.session_id: str | None = None
self._next_id = 1
self._pending: dict[int, queue.Queue[dict[str, Any]]] = {}
self._interactive: dict[int, queue.Queue[str | None]] = {}
self._lock = threading.RLock()
self._write_lock = threading.Lock()
self._reader: threading.Thread | None = None
self._closed = False
def start(self) -> None:
with self._lock:
if self.process and self.process.poll() is None:
return
env = os.environ.copy()
env["LINEUP_ADAPTER_CHILD"] = "1"
self.process = subprocess.Popen(
[self.hermes_bin, "acp"], cwd=self.cwd, env=env, text=True,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
bufsize=1,
)
self._closed = False
self._reader = threading.Thread(target=self._read_loop, name="lineup-hermes-acp", daemon=True)
self._reader.start()
initialized = self.request("initialize", {
"protocolVersion": 1,
"clientCapabilities": {"fs": {"readTextFile": False, "writeTextFile": False}, "terminal": False, "auth": {"terminal": False}},
"clientInfo": {"name": "lineup-hermes-adapter", "version": "1"},
})
if not isinstance(initialized, dict):
raise ACPError("Hermes ACP returned an invalid initialize response")
def open_session(self) -> str:
self.start()
if self.session_id:
return self.session_id
response = self.request("session/new", {"cwd": self.cwd, "mcpServers": []})
session_id = response.get("sessionId") if isinstance(response, dict) else None
if not isinstance(session_id, str) or not session_id:
raise ACPError("Hermes ACP did not return a session id")
self.session_id = session_id
return session_id
def prompt_async(self, text: str, on_done: Callable[[dict[str, Any] | None, Exception | None], None]) -> None:
session_id = self.session_id or self.open_session()
def run() -> None:
try:
result = self.request("session/prompt", {
"sessionId": session_id,
"messageId": str(uuid.uuid4()),
"prompt": [{"type": "text", "text": text}],
}, timeout=1_800.0)
except Exception as exc: # callback creates the user-safe terminal outcome
on_done(None, exc)
else:
on_done(result if isinstance(result, dict) else None, None)
threading.Thread(target=run, name="lineup-hermes-acp-prompt", daemon=True).start()
def request(self, method: str, params: dict[str, Any], timeout: float = 30.0) -> dict[str, Any]:
with self._lock:
process = self.process
if not process or process.poll() is not None or not process.stdin:
raise ACPError("Hermes ACP is not running")
request_id = self._next_id
self._next_id += 1
result_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1)
self._pending[request_id] = result_queue
self._send({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params})
try:
response = result_queue.get(timeout=timeout)
except queue.Empty as exc:
with self._lock:
self._pending.pop(request_id, None)
raise ACPError(f"Hermes ACP request timed out: {method}") from exc
if "error" in response:
error = response.get("error")
message = error.get("message") if isinstance(error, dict) else "unknown ACP error"
raise ACPError(f"Hermes ACP {method} failed: {message}")
result = response.get("result")
return result if isinstance(result, dict) else {}
def close(self) -> None:
with self._lock:
self._closed = True
process = self.process
self.process = None
self.session_id = None
pending = list(self._pending.values())
self._pending.clear()
interactive = list(self._interactive.values())
self._interactive.clear()
for response in pending:
response.put({"error": {"message": "ACP client closed"}})
for response in interactive:
response.put(None)
if process and process.poll() is None:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
def _send(self, message: dict[str, Any]) -> None:
encoded = json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n"
with self._write_lock:
process = self.process
if not process or not process.stdin:
raise ACPError("Hermes ACP stdin is unavailable")
try:
process.stdin.write(encoded)
process.stdin.flush()
except OSError as exc:
raise ACPError("Unable to write to Hermes ACP") from exc
def _read_loop(self) -> None:
error: Exception | None = None
try:
process = self.process
if not process or not process.stdout:
return
for line in process.stdout:
try:
message = json.loads(line)
except (TypeError, ValueError):
LOG.warning("discarded malformed Hermes ACP frame")
continue
if not isinstance(message, dict):
continue
if "id" in message and "method" not in message:
self._resolve_response(message)
elif message.get("method") == "session/update":
params = message.get("params")
if isinstance(params, dict):
# Keep a diagnostic signal for the M1 execution-summary
# bridge without recording ACP tool titles, arguments,
# paths, commands, outputs, or model content. This
# distinguishes an absent lifecycle event from an
# unrecognised event shape in production safely.
update = params.get("update")
if isinstance(update, dict) and update.get("sessionUpdate") in {"tool_call", "tool_call_update"}:
status = update.get("status")
safe_status = status if status in {"pending", "in_progress", "completed", "failed"} else "other"
raw_input = update.get("rawInput")
safe_update_keys = sorted(
key for key in update
if isinstance(key, str) and len(key) <= 48 and key.replace("_", "").isalnum()
)[:16]
safe_raw_input_keys = sorted(
key for key in raw_input
if isinstance(raw_input, dict) and isinstance(key, str) and len(key) <= 48 and key.replace("_", "").isalnum()
)[:16] if isinstance(raw_input, dict) else []
LOG.info(
"ACP tool lifecycle event=%s status=%s status_type=%s public_projection=%s update_keys=%s raw_input_type=%s raw_input_keys=%s",
update["sessionUpdate"], safe_status, type(status).__name__, public_tool_update(update) is not None,
safe_update_keys, type(raw_input).__name__, safe_raw_input_keys,
)
self.on_update(params)
elif "id" in message and "method" in message:
self._handle_server_request(message)
except Exception as exc:
error = exc
LOG.exception("Hermes ACP reader stopped")
finally:
if not self._closed:
self.on_closed(error or ACPError("Hermes ACP exited"))
def _resolve_response(self, message: dict[str, Any]) -> None:
request_id = message.get("id")
if not isinstance(request_id, int):
return
with self._lock:
pending = self._pending.pop(request_id, None)
if pending:
pending.put(message)
def respond_permission(self, request_id: int, option_id: str | None) -> bool:
with self._lock:
pending = self._interactive.get(request_id)
if not pending:
return False
pending.put(option_id)
return True
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
return self.respond_permission(request_id, outcome)
@staticmethod
def _parse_permission_request(message: dict[str, Any]) -> ACPInteractiveRequest | None:
request_id = message.get("id")
params = message.get("params")
if not isinstance(request_id, int) or not isinstance(params, dict):
return None
session_id = params.get("sessionId")
tool_call = params.get("toolCall")
options = params.get("options")
if not isinstance(session_id, str) or not session_id or not isinstance(tool_call, dict) or not isinstance(options, list):
return None
title = tool_call.get("title") if isinstance(tool_call.get("title"), str) else ""
if len(title) > 200:
title = title[:200]
prompt = title
content = tool_call.get("content")
if isinstance(content, list):
for block in content:
if not isinstance(block, dict):
continue
content_block = block.get("content")
if isinstance(content_block, dict) and content_block.get("type") == "text" and isinstance(content_block.get("text"), str):
prompt = content_block["text"][:2_000]
break
if not prompt:
prompt = "Hermes 请求你确认是否允许本次操作。"
projected: list[ACPInteractiveOption] = []
seen_action_ids: set[str] = set()
seen_option_ids: set[str] = set()
for index, item in enumerate(options, start=1):
if not isinstance(item, dict):
return None
option_id = item.get("optionId")
label = item.get("name")
if not _is_identifier(option_id) or option_id in seen_option_ids:
return None
if not isinstance(label, str) or not label.strip() or len(label.strip()) > 200:
return None
action_id = f"action_{index:02d}"
if action_id in seen_action_ids:
return None
seen_action_ids.add(action_id)
seen_option_ids.add(option_id)
projected.append(ACPInteractiveOption(action_id=action_id, option_id=option_id, label=label.strip()))
if not projected:
return None
return ACPInteractiveRequest(
kind="exec_approval",
request_id=request_id,
session_id=session_id,
title=title or "Hermes 权限请求",
prompt=prompt,
options=tuple(projected),
)
def _handle_server_request(self, message: dict[str, Any]) -> None:
request_id = message.get("id")
if not isinstance(request_id, int):
return
if message.get("method") == "session/request_permission":
request = self._parse_permission_request(message)
if request is None:
self._send({"jsonrpc": "2.0", "id": request_id, "result": {"outcome": {"outcome": "cancelled"}}})
return
pending: queue.Queue[str | None] = queue.Queue(maxsize=1)
with self._lock:
self._interactive[request_id] = pending
try:
self.on_request(request)
try:
selected = pending.get(timeout=900.0)
except queue.Empty:
selected = None
finally:
with self._lock:
self._interactive.pop(request_id, None)
if isinstance(selected, str):
self._send({
"jsonrpc": "2.0",
"id": request_id,
"result": {"outcome": {"optionId": selected, "outcome": "selected"}},
})
else:
self._send({"jsonrpc": "2.0", "id": request_id, "result": {"outcome": {"outcome": "cancelled"}}})
return
self._send({"jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": "Unsupported ACP client method"}})
+826
View File
@@ -0,0 +1,826 @@
"""Supervised bridge between LineUp app server messages and real Hermes CLI."""
from __future__ import annotations
import argparse
import hashlib
import json
import logging
import os
import signal
import uuid
import subprocess
import sys
import threading
import queue
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from .protocol import (
TOOL_CALL,
agent_output_message,
encode,
envelope,
execution_message,
error_message,
extract_user_input,
parse_client_inventory,
parse_agent_tool_invoke,
parse_agent_tool_call,
parse_agent_app_call,
parse_app_result,
parse_tool_response,
status_message,
tool_result_message,
validate_tool_response,
)
from .acp_client import ACPInteractiveRequest, ACPToolUpdate, HermesACPClient, public_tool_update
from .state import StateStore
LOG = logging.getLogger("lineup.adapter")
_BACKGROUND: threading.Thread | None = None
@dataclass(frozen=True)
class AdapterConfig:
appserver_url: str
agent_uid: str
channel_id: str
channel_type: int
poll_interval: float
request_timeout: float
hermes_timeout: float
hermes_bin: str
hermes_cwd: str
state_path: str
hermes_state_path: str
agent_secret: str
ignored_senders: frozenset[str]
replay_history: bool
@classmethod
def from_env(cls) -> "AdapterConfig":
root = Path(__file__).resolve().parent
return cls(
appserver_url=os.environ.get("LINEUP_APPSERVER_URL", "http://127.0.0.1:8090").rstrip("/"),
agent_uid=os.environ.get("LINEUP_AGENT_UID", "agent_hermes_main"),
channel_id=os.environ.get("LINEUP_AGENT_CHANNEL_ID", "agent_default_channel"),
channel_type=int(os.environ.get("LINEUP_CHANNEL_TYPE", "2")),
poll_interval=float(os.environ.get("LINEUP_POLL_INTERVAL", "2")),
request_timeout=float(os.environ.get("LINEUP_REQUEST_TIMEOUT_SECONDS", "10")),
hermes_timeout=float(os.environ.get("LINEUP_HERMES_TIMEOUT_SECONDS", "120")),
hermes_bin=os.environ.get("LINEUP_HERMES_BIN", "hermes"),
hermes_cwd=os.environ.get("LINEUP_HERMES_CWD", str(root)),
state_path=os.environ.get("LINEUP_STATE_PATH", str(root / "state.sqlite3")),
hermes_state_path=os.environ.get("LINEUP_HERMES_STATE_PATH", str(Path.home() / ".hermes" / "state.db")),
agent_secret=os.environ.get("LINEUP_AGENT_SHARED_SECRET", ""),
ignored_senders=frozenset(
value.strip()
for value in os.environ.get("LINEUP_IGNORED_SENDER_UIDS", "agent_default").split(",")
if value.strip()
),
replay_history=os.environ.get("LINEUP_REPLAY_HISTORY", "").lower() in {"1", "true", "yes"},
)
class ApiError(RuntimeError):
pass
class LineUpApi:
def __init__(self, config: AdapterConfig):
self.config = config
def request(self, method: str, path: str, body: dict[str, Any] | None = None, headers: dict[str, str] | None = None) -> Any:
data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
request = urllib.request.Request(
self.config.appserver_url + path,
data=data,
method=method,
headers={"Content-Type": "application/json", **(headers or {})},
)
try:
with urllib.request.urlopen(request, timeout=self.config.request_timeout) as response:
raw = response.read().decode("utf-8")
except urllib.error.HTTPError as exc:
raise ApiError(f"{method} {path}: HTTP {exc.code}: {exc.read().decode('utf-8', 'replace')[:500]}") from exc
except OSError as exc:
raise ApiError(f"{method} {path}: {exc}") from exc
try:
return json.loads(raw) if raw else {}
except ValueError as exc:
raise ApiError(f"{method} {path}: invalid JSON response") from exc
def sync(self, start_message_seq: int) -> list[dict[str, Any]]:
response = self.request(
"POST",
"/messages/sync",
{
"login_uid": self.config.agent_uid,
"channel_id": self.config.channel_id,
"channel_type": self.config.channel_type,
"start_message_seq": start_message_seq + 1 if start_message_seq else 0,
"limit": 100,
},
)
messages = response.get("messages", [])
return messages if isinstance(messages, list) else []
def send(self, payload: str) -> None:
self.request(
"POST",
"/messages/send",
{
"from_uid": self.config.agent_uid,
"channel_id": self.config.channel_id,
"channel_type": self.config.channel_type,
"payload": payload,
},
)
def heartbeat(self, status: str, last_message_seq: int, detail: str = "") -> None:
if not self.config.agent_secret:
return
self.request(
"POST",
f"/agents/{self.config.agent_uid}/heartbeat",
{"status": status, "last_message_seq": last_message_seq, "detail": detail},
{"X-LineUp-Agent-Secret": self.config.agent_secret},
)
@dataclass
class RunningTurn:
message_id: str
sender_uid: str
conversation_id: str
process: subprocess.Popen[str] | None
started_at: float
execution_id: str
trace_marker: str
acp: HermesACPClient | None = None
reply_parts: list[str] = field(default_factory=list)
acp_steps: dict[str, dict[str, str]] = field(default_factory=dict)
acp_done: bool = False
acp_error: Exception | None = None
class LineUpAdapter:
def __init__(self, config: AdapterConfig | None = None):
self.config = config or AdapterConfig.from_env()
self.api = LineUpApi(self.config)
self.store = StateStore(self.config.state_path)
self.store.recover_interrupted_turns()
self.store.recover_pending_hermes_interactions()
self.running: dict[str, RunningTurn] = {}
self.acp_clients: dict[str, HermesACPClient] = {}
self.acp_events: queue.Queue[tuple[str, object, object | None]] = queue.Queue()
self.client_inventories: dict[str, dict[str, Any]] = {}
self.stop_requested = False
def close(self) -> None:
for running in list(self.running.values()):
if running.acp:
running.acp.close()
elif running.process:
self._terminate(running.process)
for client in self.acp_clients.values():
client.close()
self.acp_clients.clear()
self.store.close()
def _send_or_queue(self, payload: str) -> None:
try:
self.api.send(payload)
except ApiError as exc:
LOG.warning("message delivery failed; queued for retry: %s", exc)
self.store.enqueue_outbox(payload)
def _flush_outbox(self) -> None:
for item in self.store.outbox_items():
try:
self.api.send(item["payload"])
except ApiError as exc:
self.store.mark_outbox_failed(item["id"], str(exc))
break
else:
self.store.mark_outbox_sent(item["id"])
def _conversation_id(self, sender_uid: str, _supplied: str | None = None) -> str:
"""Return an adapter-owned session key, never a client-claimed one.
The current AppServer group transport does not expose a separately
authenticated conversation object. Binding to configured channel plus
transport sender prevents a payload from switching Hermes to another
user's ``--continue`` session. A future server-authenticated
conversation field can be added here explicitly, rather than trusting
an envelope field by default.
"""
return f"lineup:{self.config.channel_type}:{self.config.channel_id}:{sender_uid}"
def _session_name(self, conversation_id: str) -> str:
digest = hashlib.sha256(conversation_id.encode()).hexdigest()[:20]
return f"lineup-{digest}"
def _start_turn(self, row: Any) -> None:
sender_uid = row["from_uid"]
text, supplied_conversation = extract_user_input(row["payload"])
conversation_id = self._conversation_id(sender_uid, supplied_conversation)
inventory = parse_client_inventory(row["payload"])
if inventory is not None:
# Inventory announces the host's current request surface. It is
# deliberately not a user prompt and must never itself create an
# Agent reply, capability call, or authorization decision.
self.client_inventories[conversation_id] = inventory
self.store.set_inbox_state(row["message_id"], "done", "client inventory updated")
return
app_result = parse_app_result(row["payload"])
if app_result is not None:
# This is a host-normalized outcome only. It contains neither the
# original capability arguments nor native error detail, so the
# model cannot recover clipboard text, URLs, paths or Artifact
# data from the completion notification.
disposition = self.store.consume_app_result(conversation_id, app_result["call_id"], app_result["status"])
if disposition != "accepted":
self.store.set_inbox_state(row["message_id"], "done", f"app result {disposition}")
return
text = "LineUp app capability result:\n" + json.dumps(app_result, ensure_ascii=False, separators=(",", ":"))
response = parse_tool_response(row["payload"])
if response:
interaction = self.store.hermes_interaction(conversation_id, response.call_id)
if interaction:
call = self.store.tool_call(conversation_id, response.call_id)
if not call:
self.store.set_inbox_state(row["message_id"], "done", "unknown hermes interaction")
return
try:
schema = json.loads(call["response_schema"])
except (TypeError, ValueError):
schema = None
model_result = validate_tool_response(response, schema) if isinstance(schema, dict) else None
if not model_result:
self.store.record_tool_audit(conversation_id, response.call_id, response.kind, "rejected")
self.store.set_inbox_state(row["message_id"], "done", "invalid hermes interaction response")
return
disposition = self.store.consume_tool_response(conversation_id, response.call_id, response.kind)
if disposition == "expired":
self._send_or_queue(tool_result_message(
response.call_id, "expired", conversation_id=conversation_id,
sender_id=self.config.agent_uid, target_id=sender_uid,
error="该交互已过期。",
))
self.store.set_inbox_state(row["message_id"], "done", "tool response expired")
return
if disposition != "accepted":
self.store.set_inbox_state(row["message_id"], "done", f"tool response {disposition}")
return
if response.kind == "result":
self._send_or_queue(tool_result_message(
response.call_id, response.status or "completed",
conversation_id=conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid,
))
self._resolve_hermes_interaction(conversation_id, response.call_id, model_result)
self.store.set_inbox_state(row["message_id"], "done")
return
call = self.store.tool_call(conversation_id, response.call_id)
if not call:
self.store.record_tool_audit(conversation_id, response.call_id, response.kind, "rejected")
self.store.set_inbox_state(row["message_id"], "done", "unknown tool call")
return
try:
schema = json.loads(call["response_schema"])
except (TypeError, ValueError):
schema = None
model_result = validate_tool_response(response, schema) if isinstance(schema, dict) else None
if not model_result:
self.store.record_tool_audit(conversation_id, response.call_id, response.kind, "rejected")
self.store.set_inbox_state(row["message_id"], "done", "invalid tool response")
return
disposition = self.store.consume_tool_response(conversation_id, response.call_id, response.kind)
if disposition == "expired":
self._send_or_queue(tool_result_message(
response.call_id, "expired", conversation_id=conversation_id,
sender_id=self.config.agent_uid, target_id=sender_uid,
error="该交互已过期。",
))
self.store.set_inbox_state(row["message_id"], "done", "tool response expired")
return
if disposition != "accepted":
self.store.set_inbox_state(row["message_id"], "done", f"tool response {disposition}")
return
if response.kind == "result":
# The user's transport echo is deliberately not re-ingested by
# the client. Return this adapter-authoritative terminal
# projection so the original trusted card leaves `submitted`
# and refresh restores a definitive completed state.
self._send_or_queue(tool_result_message(
response.call_id, response.status or "completed",
conversation_id=conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid,
))
# This is the only form in which user interaction reaches Hermes:
# no envelope metadata, labels, DOM/UI schema, or audit data.
text = "LineUp standard interaction result:\n" + json.dumps(model_result, ensure_ascii=False, separators=(",", ":"))
if not text.strip():
self.store.set_inbox_state(row["message_id"], "done")
return
previous = self.running.get(sender_uid)
if previous:
if previous.acp:
previous.acp.close()
elif previous.process:
self._terminate(previous.process)
self.store.set_inbox_state(previous.message_id, "cancelled", "superseded by a newer user message")
current_utc = datetime.now(timezone.utc).isoformat()
execution_id = f"exec_{uuid.uuid4().hex}"
trace_marker = f"lineup-turn:{execution_id}"
inventory_context = json.dumps(self.client_inventories.get(conversation_id, {
"revision": "none", "standard_components": [], "applications": [], "tools": [], "capabilities": [],
}), ensure_ascii=False, separators=(",", ":"))
prompt = (
"You are Hermes replying through LineUp. Reply directly and concisely to the user. "
"Do not claim to have sent a message yourself; your final text will be delivered by LineUp.\n\n"
"Default to Markdown. If a standard user interaction is essential, you may instead return exactly one valid "
"JSON object with v=1, type='lineup.v1.tool.call', and a payload for exactly one of: choice, confirm, input. "
"Every tool call needs a new call_id, a future ISO-8601 expires_at, title, prompt, and only its declaration: "
"choice data MUST be exactly {\"action_group\":{\"mode\":\"single-choice\",\"actions\":[{\"id\":\"staging\",\"label\":\"Staging\"}]}} "
"(or mode button with the same nested actions array); multi-choice is not supported here and must never be used. "
"Do not make action_group a string or put actions beside it. "
"confirm data MUST be {\"confirm\":{\"approve_label\":\"确认\",\"cancel_label\":\"取消\"}}. "
"input data MUST be {\"form\":{\"fields\":[{\"id\":\"version\",\"label\":\"Version\",\"type\":\"text\",\"required\":true,\"placeholder\":\"v1.2.3\",\"pattern\":\"^v\\\\d+\\\\.\\\\d+\\\\.\\\\d+$\"}],\"submit_label\":\"提交\",\"cancel_label\":\"取消\"}}. "
"form is an object, fields is its 1-8 item array, and every field uses id (never name). "
"All tool fields MUST be nested in the outer payload object. The only allowed shape is "
"{\"v\":1,\"type\":\"lineup.v1.tool.call\",\"payload\":{\"call_id\":\"call_...\",\"tool\":\"choice\",\"title\":\"...\",\"prompt\":\"...\",\"expires_at\":\"...\",\"data\":{...}}}. "
"Do not put call_id, tool, title, prompt, expires_at, or data beside type at the outer level.\n\n"
"For Runtime Agent Tools you may instead return exactly one JSON object with v=1, type='lineup.v1.tool.invoke'. "
"Use this only for a Tool that appears in the current inventory tools list. The only allowed shape is "
"{\"v\":1,\"type\":\"lineup.v1.tool.invoke\",\"payload\":{\"call_id\":\"call_...\",\"inventory_revision\":\"inv_...\",\"app_scope\":\"pomodoro\",\"tool_id\":\"pomodoro.start\",\"input\":{}}}. "
"Do not invent instance_id, operation_id, app_session_id, sender, target, conversation_id, or any other Runtime-internal field. "
"Use the exact latest inventory revision from the host inventory.\n\n"
f"The current UTC time is {current_utc}. expires_at must be a real ISO-8601 instant at least five minutes after that time; do not guess an earlier date or time.\n\n"
"When the user explicitly asks to open or update the task dashboard, you may instead return exactly one JSON object "
"for the host-local Surface only: ui.open payload is {instance_id,app:{app_id:'lineup.task-dashboard',version:'0.1.0'},state}; "
"ui.patch is {instance_id,state}; ui.close is {instance_id}. State is JSON only and may contain title, percent, status, steps, logs. "
"Never include bundle, URL, HTML, CSS, JavaScript, source, or any extra field. Do not emit any other ui.* message, "
"sender/target metadata, or a tool.result/tool.cancel/app.result envelope.\n\n"
"The following is a host-provided client inventory, not a grant of authority. You may only request a capability which is listed "
"there; never claim it was executed. If the user explicitly asks for one declared capability, you may instead return exactly one "
"lineup.v1.app.call JSON object with payload containing exactly call_id, capability, reason, expires_at and arguments. reason is a "
"short intent summary only and MUST NOT repeat a URL, clipboard text, file path, Artifact id, or any argument value. expires_at "
"must be at least five minutes in the future. arguments must be exactly {url} for app.open_url (https URL only), {text} for "
"clipboard.write, {} for device.pick_file, or {artifact_id} for artifact.save. The host will always ask the user to allow it once "
"and may reject it; a Surface can never bypass this native confirmation.\n\n"
"For a small generated Artifact you may instead return exactly one lineup.v1.artifact.offer JSON object. Its payload may contain only "
"artifact_id, name, mime_type, size_bytes, integrity, created_at, local_ref and optional content_base64. content_base64 is allowed only "
"for a small body with exact size_bytes and an opaque local_ref; never use URL, path, HTML, script, or source fields.\n\n"
f"{inventory_context}\n\n"
f"[Internal LineUp turn marker: {trace_marker}. Do not mention this marker in your response.]\n\n"
f"LineUp user ({sender_uid}) says:\n{text}"
)
client = self.acp_clients.get(conversation_id)
if client is None:
client = HermesACPClient(
self.config.hermes_bin,
self.config.hermes_cwd,
on_update=lambda params: self.acp_events.put(("update", params, None)),
on_request=lambda request: self.acp_events.put(("request", request, client)),
on_closed=lambda error: self.acp_events.put(("closed", client, error)),
)
self.acp_clients[conversation_id] = client
try:
client.open_session()
except (OSError, RuntimeError) as exc:
self.store.set_inbox_state(row["message_id"], "failed", str(exc))
self._send_or_queue(error_message("hermes_start_failed", "无法启动 Hermes。", conversation_id=conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid))
return
running = RunningTurn(row["message_id"], sender_uid, conversation_id, None, time.monotonic(), execution_id, trace_marker, acp=client)
self.running[sender_uid] = running
self.store.set_inbox_state(row["message_id"], "running")
self._send_or_queue(status_message("thinking", conversation_id=conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid, execution_id=execution_id))
def done(result: dict[str, Any] | None, error: Exception | None) -> None:
self.acp_events.put(("done", running, error))
client.prompt_async(prompt, done)
@staticmethod
def _terminate(process: subprocess.Popen[str]) -> None:
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
def _flush_acp_events(self) -> None:
"""Apply ACP notifications on the Adapter thread, not the reader thread."""
while True:
try:
kind, subject, detail = self.acp_events.get_nowait()
except queue.Empty:
return
if kind == "done" and isinstance(subject, RunningTurn):
subject.acp_done = True
subject.acp_error = detail if isinstance(detail, Exception) else None
continue
if kind == "closed" and isinstance(subject, HermesACPClient):
for running in self.running.values():
if running.acp is subject and not running.acp_done:
running.acp_done = True
running.acp_error = detail if isinstance(detail, Exception) else RuntimeError("Hermes ACP exited")
continue
if (
kind == "request"
and isinstance(subject, ACPInteractiveRequest)
and detail is not None
and hasattr(detail, "session_id")
and (hasattr(detail, "respond_interaction") or hasattr(detail, "respond_permission"))
):
running = next((item for item in self.running.values() if item.acp is detail and detail.session_id == subject.session_id), None)
if running:
self._emit_interaction_request(running, detail, subject)
else:
self._respond_interaction(detail, subject.request_id, None)
continue
if kind != "update" or not isinstance(subject, dict):
continue
session_id = subject.get("sessionId")
update = subject.get("update")
running = next((item for item in self.running.values() if item.acp and item.acp.session_id == session_id), None)
if not running or not isinstance(update, dict):
continue
public = public_tool_update(
update,
{tool_id: step["title"] for tool_id, step in running.acp_steps.items()},
)
if public:
self._apply_acp_tool_update(running, public)
continue
if update.get("sessionUpdate") == "agent_message_chunk":
content = update.get("content")
if isinstance(content, dict) and content.get("type") == "text" and isinstance(content.get("text"), str):
running.reply_parts.append(content["text"])
def _apply_acp_tool_update(self, running: RunningTurn, update: ACPToolUpdate) -> None:
step = running.acp_steps.get(update.tool_id)
if step is None:
step = {"id": f"step_{len(running.acp_steps) + 1:02d}", "title": update.title, "status": update.status}
running.acp_steps[update.tool_id] = step
else:
step["status"] = update.status
self._send_or_queue(execution_message(
running.execution_id, "running", list(running.acp_steps.values()),
conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid,
))
@staticmethod
def _respond_interaction(client: Any, request_id: int, outcome: str | None) -> bool:
responder = getattr(client, "respond_interaction", None) or getattr(client, "respond_permission", None)
return bool(responder and responder(request_id, outcome))
def _emit_interaction_request(
self, running: RunningTurn, client: Any, request: ACPInteractiveRequest,
) -> None:
if request.kind == "clarify" and request.multi_select:
self._respond_interaction(client, request.request_id, None)
return
expires_at_ts = time.time() + 900
expires_at = datetime.fromtimestamp(expires_at_ts, tz=timezone.utc).isoformat()
call_id = f"call_{uuid.uuid4().hex}"
title = request.title or "Hermes 交互请求"
prompt = request.prompt or "Hermes 请求你补充确认。"
payload: dict[str, Any]
response_schema: dict[str, Any]
interaction_meta: dict[str, Any]
if request.kind == "update_prompt":
payload = {
"call_id": call_id,
"tool": "confirm",
"title": title,
"prompt": prompt,
"expires_at": expires_at,
"data": {"confirm": {"approve_label": "确认", "cancel_label": "取消"}},
}
response_schema = {"tool": "confirm"}
interaction_meta = {"mode": "confirm", "confirmed": "y", "cancelled": "n"}
elif request.kind == "clarify" and request.expects_text and not request.options:
field_id = "answer"
payload = {
"call_id": call_id,
"tool": "input",
"title": title,
"prompt": prompt,
"expires_at": expires_at,
"data": {
"form": {
"fields": [{"id": field_id, "label": "你的回答", "type": "text", "required": True}],
"submit_label": "提交",
"cancel_label": "取消",
}
},
}
response_schema = {"tool": "input", "fields": {field_id: {"type": "text", "required": True, "min_length": None, "max_length": None}}}
interaction_meta = {"mode": "input", "field_id": field_id}
else:
actions = [{"id": option.action_id, "label": option.label} for option in request.options]
mapping = {option.action_id: option.option_id for option in request.options}
other_action_id: str | None = None
if request.kind == "clarify" and request.allow_other:
other_action_id = f"action_{len(actions) + 1:02d}"
actions.append({"id": other_action_id, "label": "其他"})
mapping[other_action_id] = "__other__"
payload = {
"call_id": call_id,
"tool": "choice",
"title": title,
"prompt": prompt,
"expires_at": expires_at,
"data": {"action_group": {"mode": "single-choice", "actions": actions}},
}
response_schema = {"tool": "choice", "mode": "single-choice", "action_ids": sorted(action["id"] for action in actions)}
interaction_meta = {"mode": "choice", "mapping": mapping}
if other_action_id:
interaction_meta["other_action_id"] = other_action_id
disposition = self.store.register_tool_call(running.conversation_id, call_id, payload["tool"], expires_at_ts, response_schema)
if disposition != "accepted":
self._respond_interaction(client, request.request_id, None)
return
interaction_disposition = self.store.register_hermes_interaction(
running.conversation_id,
call_id,
kind=request.kind,
session_id=request.session_id,
acp_request_id=request.request_id,
option_map=interaction_meta,
)
if interaction_disposition != "accepted":
self.store.complete_hermes_interaction(running.conversation_id, call_id, "duplicate")
self._respond_interaction(client, request.request_id, None)
return
self._send_or_queue(encode(envelope(
TOOL_CALL,
payload,
conversation_id=running.conversation_id,
sender_id=self.config.agent_uid,
target_id=running.sender_uid,
)))
def _resolve_hermes_interaction(self, conversation_id: str, call_id: str, model_result: dict[str, Any]) -> None:
interaction = self.store.hermes_interaction(conversation_id, call_id)
if not interaction:
return
outcome: str | None = None
state = "cancelled"
try:
metadata = json.loads(interaction["option_map"])
except (TypeError, ValueError):
metadata = {}
mode = metadata.get("mode") if isinstance(metadata, dict) else None
if mode == "choice":
mapping = metadata.get("mapping")
selected = model_result.get("action_ids")
if isinstance(mapping, dict) and isinstance(selected, list) and len(selected) == 1 and selected[0] in mapping:
value = mapping[selected[0]]
if interaction["kind"] == "clarify" and value == "__other__":
self.store.complete_hermes_interaction(conversation_id, call_id, "continued")
client = self.acp_clients.get(conversation_id)
if client:
followup = ACPInteractiveRequest(
kind="clarify",
request_id=int(interaction["acp_request_id"]),
session_id=str(interaction["session_id"]),
title="补充说明",
prompt="请输入其他说明。",
expects_text=True,
)
running = next((item for item in self.running.values() if item.conversation_id == conversation_id and item.acp is client), None)
if running:
self._emit_interaction_request(running, client, followup)
return
outcome = value if isinstance(value, str) else None
state = "resolved"
elif mode == "confirm":
if model_result.get("outcome") == "confirmed":
outcome = metadata.get("confirmed")
state = "resolved"
elif model_result.get("outcome") == "cancelled":
outcome = metadata.get("cancelled")
state = "cancelled"
elif mode == "input":
values = model_result.get("values")
field_id = metadata.get("field_id")
if isinstance(values, dict) and isinstance(field_id, str) and isinstance(values.get(field_id), str):
outcome = values[field_id]
state = "resolved"
client = self.acp_clients.get(conversation_id)
if client and isinstance(interaction["acp_request_id"], int):
self._respond_interaction(client, interaction["acp_request_id"], outcome)
self.store.complete_hermes_interaction(conversation_id, call_id, state)
def _finish_acp(self, running: RunningTurn, failure: Exception | None = None) -> None:
self.running.pop(running.sender_uid, None)
if failure:
for step in running.acp_steps.values():
if step["status"] == "running":
step["status"] = "failed"
self.store.set_inbox_state(running.message_id, "failed", str(failure))
self._send_or_queue(execution_message(running.execution_id, "failed", list(running.acp_steps.values()), conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid))
self._send_or_queue(error_message("hermes_acp_failed", "Hermes 未能完成本次请求。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid))
return
reply = "".join(running.reply_parts).strip()
if not reply:
self._finish_acp(running, RuntimeError("empty Hermes ACP response"))
return
outgoing = agent_output_message(reply, conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid, execution_id=running.execution_id, inventory=self.client_inventories.get(running.conversation_id))
tool_invoke = parse_agent_tool_invoke(reply, self.client_inventories.get(running.conversation_id))
tool_call = parse_agent_tool_call(reply)
if tool_invoke and tool_call:
outgoing = error_message("conflicting_tool_output", "Agent 同时返回了工具调用与交互请求,请只保留一种。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid)
elif tool_call:
canonical, response_schema, expires_at = tool_call
disposition = self.store.register_tool_call(running.conversation_id, canonical["call_id"], canonical["tool"], expires_at, response_schema)
if disposition != "accepted":
outgoing = error_message("duplicate_tool_call", "Agent 重复创建了同一交互,请重新发起。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid)
app_call = parse_agent_app_call(reply, self.client_inventories.get(running.conversation_id))
if app_call:
canonical, expires_at = app_call
if self.store.register_app_call(running.conversation_id, canonical["call_id"], canonical["capability"], expires_at) != "accepted":
outgoing = error_message("duplicate_app_call", "Agent 重复创建了同一 App 操作,请重新发起。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid)
self._send_or_queue(execution_message(running.execution_id, "completed", list(running.acp_steps.values()), conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid))
self._send_or_queue(outgoing)
self._send_or_queue(status_message("idle", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=running.sender_uid, execution_id=running.execution_id))
self.store.set_inbox_state(running.message_id, "done")
def _reap_turns(self) -> None:
for sender_uid, running in list(self.running.items()):
timed_out = time.monotonic() - running.started_at > self.config.hermes_timeout
if running.acp:
if timed_out:
running.acp.close()
self._finish_acp(running, RuntimeError("Hermes ACP response timeout"))
elif running.acp_done:
self._finish_acp(running, running.acp_error)
continue
if running.process.poll() is None and not timed_out:
continue
if timed_out:
self._terminate(running.process)
stdout, stderr = running.process.communicate()
del self.running[sender_uid]
if timed_out:
self.store.set_inbox_state(running.message_id, "failed", "Hermes response timeout")
# A status "thinking" begins a client-side execution card.
# Always close that exact id on terminal failure as well: an
# error envelope has no execution_id, so relying on it alone
# leaves a stale running card after a timeout.
self._send_or_queue(execution_message(
running.execution_id, "failed", [],
conversation_id=running.conversation_id,
sender_id=self.config.agent_uid,
target_id=sender_uid,
))
self._send_or_queue(error_message("hermes_timeout", "Hermes 响应超时,请重试。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid))
continue
if running.process.returncode != 0:
self.store.set_inbox_state(running.message_id, "failed", stderr.strip()[:1000])
LOG.error("Hermes exited %s: %s", running.process.returncode, stderr.strip())
self._send_or_queue(execution_message(
running.execution_id, "failed", [],
conversation_id=running.conversation_id,
sender_id=self.config.agent_uid,
target_id=sender_uid,
))
self._send_or_queue(error_message("hermes_failed", "Hermes 未能完成本次请求。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid))
continue
reply = stdout.strip()
if not reply:
self.store.set_inbox_state(running.message_id, "failed", "empty Hermes response")
self._send_or_queue(execution_message(
running.execution_id, "failed", [],
conversation_id=running.conversation_id,
sender_id=self.config.agent_uid,
target_id=sender_uid,
))
self._send_or_queue(error_message("hermes_empty_response", "Hermes 没有返回内容。", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid))
continue
outgoing = agent_output_message(reply, conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid, execution_id=running.execution_id, inventory=self.client_inventories.get(running.conversation_id))
tool_invoke = parse_agent_tool_invoke(reply, self.client_inventories.get(running.conversation_id))
tool_call = parse_agent_tool_call(reply)
if tool_invoke and tool_call:
outgoing = error_message(
"conflicting_tool_output", "Agent 同时返回了工具调用与交互请求,请只保留一种。",
conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid,
)
elif tool_call:
canonical, response_schema, expires_at = tool_call
disposition = self.store.register_tool_call(
running.conversation_id, canonical["call_id"], canonical["tool"], expires_at, response_schema
)
if disposition != "accepted":
LOG.warning("suppressed duplicate tool call %s in %s", canonical["call_id"], running.conversation_id)
outgoing = error_message(
"duplicate_tool_call", "Agent 重复创建了同一交互,请重新发起。",
conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid,
)
app_call = parse_agent_app_call(reply, self.client_inventories.get(running.conversation_id))
if app_call:
canonical, expires_at = app_call
if self.store.register_app_call(running.conversation_id, canonical["call_id"], canonical["capability"], expires_at) != "accepted":
outgoing = error_message(
"duplicate_app_call", "Agent 重复创建了同一 App 操作,请重新发起。",
conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid,
)
self._send_or_queue(outgoing)
self._send_or_queue(status_message("idle", conversation_id=running.conversation_id, sender_id=self.config.agent_uid, target_id=sender_uid, execution_id=running.execution_id))
self.store.set_inbox_state(running.message_id, "done")
def run_once(self) -> None:
self._flush_outbox()
self._flush_acp_events()
self._reap_turns()
cursor = self.store.cursor()
try:
messages = self.api.sync(cursor)
except ApiError as exc:
LOG.warning("sync failed: %s", exc)
return
# A brand-new worker must not turn the entire retained IM history into
# fresh Hermes prompts. Operators can explicitly opt into replay for a
# controlled migration or recovery run.
if not self.store.is_bootstrapped() and not self.config.replay_history:
newest = max((int(message.get("message_seq") or 0) for message in messages), default=cursor)
if newest > cursor:
self.store.set_cursor(newest)
self.store.set_bootstrapped()
LOG.info("adapter bootstrap complete at message sequence %s; retained history was not replayed", newest)
return
for message in messages:
seq = int(message.get("message_seq") or 0)
if message.get("from_uid") not in {self.config.agent_uid, *self.config.ignored_senders}:
self.store.claim_inbox(message)
if seq > cursor:
cursor = seq
self.store.set_cursor(cursor)
self.store.set_bootstrapped()
for row in self.store.pending():
if row["state"] == "pending":
self._start_turn(row)
try:
self.api.heartbeat("running", self.store.cursor(), f"active_turns={len(self.running)}")
except ApiError as exc:
LOG.warning("heartbeat failed: %s", exc)
def run_forever(self) -> None:
while not self.stop_requested:
self.run_once()
time.sleep(self.config.poll_interval)
def start_background() -> None:
global _BACKGROUND
if _BACKGROUND and _BACKGROUND.is_alive():
return
def runner() -> None:
adapter = LineUpAdapter()
try:
adapter.run_forever()
finally:
adapter.close()
_BACKGROUND = threading.Thread(target=runner, name="lineup-adapter", daemon=True)
_BACKGROUND.start()
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="LineUp Hermes Adapter")
parser.add_argument("--once", action="store_true", help="process one poll cycle and exit")
parser.add_argument("--check", action="store_true", help="validate configuration and print persistent queue state")
args = parser.parse_args(argv)
logging.basicConfig(level=os.environ.get("LINEUP_LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(name)s: %(message)s")
adapter = LineUpAdapter()
try:
if args.check:
print(adapter.store.status_summary())
return 0
if args.once:
adapter.run_once()
return 0
def stop(_signum: int, _frame: Any) -> None:
LOG.info("received termination signal; stopping LineUp Adapter")
adapter.stop_requested = True
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
adapter.run_forever()
return 0
finally:
adapter.close()
if __name__ == "__main__":
raise SystemExit(main())
+108
View File
@@ -0,0 +1,108 @@
"""Safe public operation summaries derived from Hermes' local tool ledger.
This module is deliberately a one-way privacy filter. It reads only enough
metadata to establish that a tool operation happened, then returns a bounded
allowlist of human labels. Raw commands, arguments, paths, URLs, outputs,
errors, and model reasoning never leave this process.
"""
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
from typing import Any
MAX_STEPS = 12
def extract_public_execution_steps(state_db_path: str, marker: str) -> list[dict[str, str]]:
"""Return public labels for the exact Hermes turn tagged by *marker*.
Failure to open or recognize the best-effort local ledger is safe: callers
receive no trace rather than an invented one.
"""
if not marker or not Path(state_db_path).is_file():
return []
try:
connection = sqlite3.connect(f"file:{Path(state_db_path).resolve()}?mode=ro", uri=True)
connection.row_factory = sqlite3.Row
try:
marker_row = connection.execute(
"SELECT session_id, timestamp FROM messages WHERE content LIKE ? ORDER BY timestamp DESC LIMIT 1",
(f"%{marker}%",),
).fetchone()
if not marker_row or not marker_row["session_id"] or marker_row["timestamp"] is None:
return []
rows = connection.execute(
"SELECT role, tool_name, tool_calls, content FROM messages "
"WHERE session_id = ? AND timestamp >= ? ORDER BY timestamp ASC, id ASC",
(marker_row["session_id"], marker_row["timestamp"]),
).fetchall()
finally:
connection.close()
except (sqlite3.Error, OSError, ValueError, KeyError, TypeError):
return []
labels: list[str] = []
for row in rows:
if row["role"] not in {"assistant", "tool"}:
continue
for tool_name, opaque in _tool_operations(row):
label = _public_label(tool_name, opaque)
if label and label not in labels:
labels.append(label)
if len(labels) >= MAX_STEPS:
break
if len(labels) >= MAX_STEPS:
break
return [{"id": f"step_{index:02d}", "title": title, "status": "completed"} for index, title in enumerate(labels, start=1)]
def _tool_operations(row: sqlite3.Row) -> list[tuple[str, str]]:
"""Read tool identity/opaque metadata only; never return it to callers."""
result: list[tuple[str, str]] = []
tool_name = row["tool_name"]
if isinstance(tool_name, str) and tool_name:
result.append((tool_name, _text(row["content"])))
raw_calls = row["tool_calls"]
if not isinstance(raw_calls, str) or not raw_calls:
return result
try:
decoded: Any = json.loads(raw_calls)
except (TypeError, ValueError):
return result
calls = decoded if isinstance(decoded, list) else [decoded]
for call in calls:
if not isinstance(call, dict):
continue
name = call.get("name") or call.get("tool_name") or (call.get("function") or {}).get("name")
if isinstance(name, str) and name:
# This stays opaque: it is examined solely by fixed predicates.
result.append((name, json.dumps(call, ensure_ascii=False, separators=(",", ":"))))
return result
def _public_label(tool_name: str, opaque: str) -> str | None:
name = tool_name.lower()
lower = opaque.lower()
if name in {"web_search", "search_web"}:
return "查询公开资料"
if name in {"web_extract", "web_fetch", "read_web"}:
return "读取公开网页内容"
if name in {"search_files", "find_files"}:
return "检索相关文件"
if name in {"read_file", "read_files"}:
return "读取相关配置"
if name in {"terminal", "shell", "command"}:
# These labels are intentionally specific only for a narrow allowlist.
if "openapi.json" in lower:
return "获取服务接口说明"
if "/kline" in lower:
return "查询行情接口"
if "/health" in lower and ("qmtbridge" in lower or "qmt" in lower):
return "检查 qmtbridge health"
return None
def _text(value: Any) -> str:
return value if isinstance(value, str) else ""
+23
View File
@@ -0,0 +1,23 @@
manifest_version: 1
name: lineup
label: LineUp
kind: standalone
version: 0.1.0
description: >
Connects a supervised LineUp Adapter to the local Hermes runtime. The adapter
consumes LineUp messages through the LineUp app server, preserves per-user
Hermes sessions, and sends versioned LineUp protocol envelopes back.
author: LineUp
optional_env:
- name: LINEUP_APPSERVER_URL
description: "LineUp app server base URL (default: http://127.0.0.1:8090)"
prompt: "LineUp app server URL"
password: false
- name: LINEUP_AGENT_UID
description: "Stable WuKongIM/LineUp identity for this Hermes instance"
prompt: "LineUp agent UID"
password: false
- name: LINEUP_AGENT_SHARED_SECRET
description: "Agent heartbeat credential configured on the LineUp app server"
prompt: "LineUp agent shared secret"
password: true
+943
View File
@@ -0,0 +1,943 @@
"""Versioned LineUp envelope helpers with a narrow legacy-input bridge."""
from __future__ import annotations
import json
import re
import uuid
import base64
import binascii
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
PROTOCOL_VERSION = 1
TEXT = "lineup.v1.text"
STATUS = "lineup.v1.agent.status"
EXECUTION = "lineup.v1.agent.execution"
PROGRESS = "lineup.v1.agent.progress"
TOOL_CALL = "lineup.v1.tool.call"
TOOL_INVOKE = "lineup.v1.tool.invoke"
TOOL_RESULT = "lineup.v1.tool.result"
TOOL_CANCEL = "lineup.v1.tool.cancel"
ERROR = "lineup.v1.error"
UI_OPEN = "lineup.v1.ui.open"
UI_PATCH = "lineup.v1.ui.patch"
UI_CLOSE = "lineup.v1.ui.close"
UI_EVENT = "lineup.v1.ui.event"
CLIENT_INVENTORY = "lineup.v1.client.inventory"
APP_LIST = "lineup.v1.app.list"
APP_CALL = "lineup.v1.app.call"
APP_RESULT = "lineup.v1.app.result"
ARTIFACT_OFFER = "lineup.v1.artifact.offer"
# M2 permits one host-local, bundle-free Surface lifecycle. Capability calls
# stay outside this set: a Surface is presentation-only and never grants a
# device, DOM, network, or native bridge capability.
AGENT_OUTPUT_TYPES = frozenset({TEXT, STATUS, PROGRESS, TOOL_CALL, TOOL_INVOKE, ERROR, UI_OPEN, UI_PATCH, UI_CLOSE, ARTIFACT_OFFER})
_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$")
_TOOL_KINDS = frozenset({"choice", "confirm", "input"})
_AGENT_TOOL_HANDLING = frozenset({"direct", "interactive", "launch", "foreground", "operation"})
_AGENT_TOOL_DELIVERY = frozenset({"miniapp_sdk", "runtime"})
_AGENT_TOOL_ACTIVATION = frozenset({"activation_not_required", "foreground_required", "app_ready"})
_ACTION_MODES = frozenset({"single-choice", "button"})
_FIELD_TYPES = frozenset({"text", "textarea", "number"})
_RESULT_STATUSES = frozenset({"completed", "failed", "cancelled", "expired"})
_TASK_DASHBOARD_APP = "lineup.task-dashboard"
_TASK_DASHBOARD_VERSION = "0.1.0"
_SURFACE_OVERRIDE_FIELDS = frozenset({"bundle", "bundle_id", "bundle_url", "url", "html", "css", "javascript", "js", "script", "source"})
_INVENTORY_CAPABILITIES = frozenset({"app.open_url", "clipboard.write", "device.pick_file", "artifact.save"})
_APP_CALL_ARGUMENTS = {
"app.open_url": frozenset({"url"}),
"clipboard.write": frozenset({"text"}),
"device.pick_file": frozenset(),
"artifact.save": frozenset({"artifact_id"}),
}
_APP_RESULT_STATUSES = frozenset({"completed", "failed", "rejected", "cancelled", "expired", "unsupported"})
_SENSITIVE_REASON = re.compile(r"https?://|(?:[A-Za-z]:)?[\\/]")
_ARTIFACT_INTEGRITY = frozenset({"verified", "unverified", "failed"})
_OPAQUE_REF = re.compile(r"^[A-Za-z0-9_.:-]{1,128}$")
_BASE64 = re.compile(r"^[A-Za-z0-9+/]*={0,2}$")
@dataclass(frozen=True)
class ToolResponse:
"""A validated client response, deliberately without envelope metadata."""
kind: str
call_id: str
status: str | None = None
result: dict[str, Any] | None = None
reason: str | None = None
def _identifier(value: Any) -> str | None:
return value if isinstance(value, str) and _IDENTIFIER.fullmatch(value) else None
def _bounded_text(value: Any, maximum: int, *, allow_empty: bool = True) -> str | None:
if not isinstance(value, str):
return None
normalized = value.strip()
if len(normalized) > maximum or (not allow_empty and not normalized):
return None
return normalized
def _timestamp(value: Any) -> tuple[str, float] | None:
if not isinstance(value, str):
return None
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
return None
return value, parsed.timestamp()
def _json_object(value: Any) -> dict[str, Any] | None:
return value if isinstance(value, dict) else None
def parse_client_inventory(raw_payload: str) -> dict[str, Any] | None:
"""Accept a bounded declarative inventory without granting authority.
The Adapter stores this only as prompt context for the same transport
sender. It is not an Agent turn, does not authorize an app.call, and is
reduced to fields that cannot contain a bundle, path, token or user data.
"""
try:
envelope_value = json.loads(raw_payload)
except (TypeError, ValueError):
return None
if not isinstance(envelope_value, dict) or envelope_value.get("type") != CLIENT_INVENTORY:
return None
payload = _json_object(envelope_value.get("payload"))
if not payload or set(payload) not in ({"revision", "standard_components", "applications", "capabilities"}, {"revision", "standard_components", "applications", "tools", "capabilities"}):
return None
revision = _bounded_text(payload.get("revision"), 128, allow_empty=False)
components = payload.get("standard_components")
applications = payload.get("applications")
tools = payload.get("tools", [])
capabilities = payload.get("capabilities")
if not revision or not isinstance(components, list) or not isinstance(applications, list) or not isinstance(tools, list) or not isinstance(capabilities, list):
return None
if len(components) > 8 or len(applications) > 16 or len(tools) > 64 or len(capabilities) > 16:
return None
canonical_components: list[dict[str, str]] = []
for value in components:
item = _json_object(value)
component_id = _bounded_text(item.get("id"), 64, allow_empty=False) if item else None
version = _bounded_text(item.get("version"), 32, allow_empty=False) if item else None
if not component_id or not version or set(item) != {"id", "version"}:
return None
canonical_components.append({"id": component_id, "version": version})
canonical_applications: list[dict[str, Any]] = []
for value in applications:
item = _json_object(value)
app_id = _bounded_text(item.get("id"), 96, allow_empty=False) if item else None
version = _bounded_text(item.get("version"), 64, allow_empty=False) if item else None
surfaces = item.get("surfaces") if item else None
if not app_id or not version or not isinstance(surfaces, list) or len(surfaces) > 8 or set(item) != {"id", "version", "surfaces"}:
return None
canonical_surfaces: list[dict[str, Any]] = []
for surface in surfaces:
definition = _json_object(surface)
surface_id = _bounded_text(definition.get("id"), 64, allow_empty=False) if definition else None
events = definition.get("events") if definition else None
if not surface_id or not isinstance(events, list) or len(events) > 16 or set(definition) != {"id", "events"} or not all(_identifier(event) for event in events):
return None
canonical_surfaces.append({"id": surface_id, "events": list(events)})
canonical_applications.append({"id": app_id, "version": version, "surfaces": canonical_surfaces})
canonical_tools: list[dict[str, Any]] = []
for value in tools:
item = _json_object(value)
app_scope = _bounded_text(item.get("app_scope"), 96, allow_empty=False) if item else None
tool_id = _bounded_text(item.get("tool_id"), 128, allow_empty=False) if item else None
description = _bounded_text(item.get("description"), 500, allow_empty=False) if item else None
input_schema = _json_object(item.get("input_schema")) if item else None
handling = item.get("handling") if item else None
delivery = item.get("delivery") if item else None
activation_requirement = item.get("activation_requirement") if item else None
if (not app_scope or not tool_id or not description or not input_schema
or handling not in _AGENT_TOOL_HANDLING
or delivery not in _AGENT_TOOL_DELIVERY
or activation_requirement not in _AGENT_TOOL_ACTIVATION):
return None
if set(item) != {"app_scope", "tool_id", "description", "input_schema", "handling", "delivery", "activation_requirement"}:
return None
canonical_tools.append({
"app_scope": app_scope,
"tool_id": tool_id,
"description": description,
"input_schema": input_schema,
"handling": handling,
"delivery": delivery,
"activation_requirement": activation_requirement,
})
canonical_capabilities: list[dict[str, Any]] = []
for value in capabilities:
item = _json_object(value)
name = item.get("name") if item else None
version = _bounded_text(item.get("version"), 32, allow_empty=False) if item else None
risk = item.get("risk") if item else None
schema = _json_object(item.get("input_schema")) if item else None
foreground_required = item.get("foreground_required") if item else None
if name not in _INVENTORY_CAPABILITIES or not version or risk not in {"display_only", "user_confirmation", "system_permission", "restricted"} or not schema or foreground_required is not True:
return None
if set(item) != {"name", "version", "risk", "input_schema", "foreground_required"} or set(schema) != {"type", "required", "properties"}:
return None
canonical_capabilities.append({"name": name, "version": version, "risk": risk, "input_schema": schema, "foreground_required": True})
return {"revision": revision, "standard_components": canonical_components, "applications": canonical_applications, "tools": canonical_tools, "capabilities": canonical_capabilities}
def validate_tool_invoke_payload(payload: dict[str, Any], inventory: dict[str, Any] | None) -> dict[str, Any] | None:
"""Canonically admit one Runtime Agent Tool invoke for *this* inventory."""
if set(payload) != {"call_id", "inventory_revision", "app_scope", "tool_id", "input"}:
return None
call_id = _identifier(payload.get("call_id"))
inventory_revision = _bounded_text(payload.get("inventory_revision"), 128, allow_empty=False)
app_scope = _bounded_text(payload.get("app_scope"), 96, allow_empty=False)
tool_id = _bounded_text(payload.get("tool_id"), 128, allow_empty=False)
input_value = _json_object(payload.get("input"))
if not call_id or not inventory_revision or not app_scope or not tool_id or input_value is None:
return None
if not isinstance(inventory, dict) or inventory.get("revision") != inventory_revision:
return None
visible = inventory.get("tools")
if not isinstance(visible, list):
return None
if not any(
isinstance(item, dict)
and item.get("app_scope") == app_scope
and item.get("tool_id") == tool_id
for item in visible
):
return None
return {
"call_id": call_id,
"inventory_revision": inventory_revision,
"app_scope": app_scope,
"tool_id": tool_id,
"input": input_value,
}
def validate_app_call_payload(payload: dict[str, Any], inventory: dict[str, Any] | None, *, now: datetime | None = None) -> tuple[dict[str, Any], float] | None:
"""Canonically admit one Agent capability request for *this* inventory.
Inventory is still not authorization. It only narrows what the adapter
may place on the wire; the host performs its own policy, confirmation and
execution-time checks. Unknown keys and non-string nested values are
rejected rather than passed through as model-controlled UI or parameters.
"""
if set(payload) != {"call_id", "capability", "reason", "expires_at", "arguments"}:
return None
call_id = _identifier(payload.get("call_id"))
capability = payload.get("capability")
reason = _bounded_text(payload.get("reason"), 500, allow_empty=False)
expiry = _timestamp(payload.get("expires_at"))
arguments = _json_object(payload.get("arguments"))
if not call_id or capability not in _APP_CALL_ARGUMENTS or not reason or _SENSITIVE_REASON.search(reason) or not expiry or arguments is None:
return None
reference = now or datetime.now(timezone.utc)
if expiry[1] <= reference.timestamp():
return None
visible = inventory.get("capabilities") if isinstance(inventory, dict) else None
declared = {
item.get("name") for item in visible
if isinstance(item, dict) and item.get("name") in _INVENTORY_CAPABILITIES
} if isinstance(visible, list) else set()
if capability not in declared or set(arguments) != _APP_CALL_ARGUMENTS[capability]:
return None
canonical_arguments: dict[str, str] = {}
for key in _APP_CALL_ARGUMENTS[capability]:
value = _bounded_text(arguments.get(key), 8_192, allow_empty=False)
if value is None:
return None
canonical_arguments[key] = value
return ({"call_id": call_id, "capability": capability, "reason": reason, "expires_at": expiry[0], "arguments": canonical_arguments}, expiry[1])
def validate_artifact_offer_payload(payload: dict[str, Any]) -> dict[str, Any] | None:
"""Canonicalize a bounded Artifact offer without accepting a URL/path."""
allowed = {"artifact_id", "name", "mime_type", "size_bytes", "integrity", "created_at", "local_ref", "content_base64"}
if set(payload) - allowed:
return None
artifact_id = _identifier(payload.get("artifact_id"))
name = _bounded_text(payload.get("name"), 255, allow_empty=False)
mime_type = _bounded_text(payload.get("mime_type"), 128, allow_empty=False)
size = payload.get("size_bytes")
integrity = payload.get("integrity")
created = _timestamp(payload.get("created_at"))
local_ref = payload.get("local_ref")
if (not artifact_id or not name or not mime_type or not re.fullmatch(r"[A-Za-z0-9.+-]+/[A-Za-z0-9.+-]+", mime_type)
or not isinstance(size, int) or isinstance(size, bool) or size < 0 or size > 512 * 1024 * 1024
or integrity not in _ARTIFACT_INTEGRITY or not created):
return None
canonical: dict[str, Any] = {"artifact_id": artifact_id, "name": name, "mime_type": mime_type, "size_bytes": size, "integrity": integrity, "created_at": created[0]}
if local_ref is not None:
if not isinstance(local_ref, str) or not _OPAQUE_REF.fullmatch(local_ref) or "/" in local_ref or "\\" in local_ref:
return None
canonical["local_ref"] = local_ref
content = payload.get("content_base64")
if content is not None:
if not isinstance(content, str) or len(content) > 8 * 1024 * 1024 or not _BASE64.fullmatch(content) or not local_ref:
return None
try:
decoded = base64.b64decode(content, validate=True)
except (ValueError, binascii.Error):
return None
if len(decoded) != size or len(decoded) > 6 * 1024 * 1024:
return None
canonical["content_base64"] = content
return canonical
def parse_app_result(raw_payload: str) -> dict[str, Any] | None:
"""Accept the host's bounded outcome projection, not arbitrary result data."""
try:
value = json.loads(raw_payload)
except (TypeError, ValueError):
return None
if not isinstance(value, dict) or value.get("v") != PROTOCOL_VERSION or value.get("type") != APP_RESULT:
return None
payload = _json_object(value.get("payload"))
if not payload or set(payload) - {"call_id", "capability", "status", "completed", "code"}:
return None
call_id = _identifier(payload.get("call_id"))
capability = payload.get("capability")
status = payload.get("status")
if not call_id or capability not in _INVENTORY_CAPABILITIES or status not in _APP_RESULT_STATUSES:
return None
if payload.get("completed") is not None and payload.get("completed") is not True:
return None
code = payload.get("code")
if code is not None and not _bounded_text(code, 128, allow_empty=False):
return None
return {"call_id": call_id, "capability": capability, "status": status, **({"completed": True} if payload.get("completed") is True else {}), **({"code": code} if isinstance(code, str) else {})}
def validate_surface_payload(message_type: str, payload: dict[str, Any]) -> dict[str, Any] | None:
"""Canonicalize the sole M2 Surface lifecycle without accepting code."""
instance_id = _identifier(payload.get("instance_id"))
if not instance_id or any(field in payload for field in _SURFACE_OVERRIDE_FIELDS):
return None
if message_type == UI_OPEN:
app = _json_object(payload.get("app"))
state = _json_object(payload.get("state", {}))
if (not app or set(app) != {"app_id", "version"}
or app.get("app_id") != _TASK_DASHBOARD_APP or app.get("version") != _TASK_DASHBOARD_VERSION
or state is None or set(payload) - {"instance_id", "app", "state"}):
return None
if len(json.dumps(state, ensure_ascii=False).encode("utf-8")) > 32 * 1024:
return None
return {"instance_id": instance_id, "app": {"app_id": _TASK_DASHBOARD_APP, "version": _TASK_DASHBOARD_VERSION}, "state": state}
if message_type == UI_PATCH:
state = _json_object(payload.get("state"))
if state is None or set(payload) != {"instance_id", "state"} or len(json.dumps(state, ensure_ascii=False).encode("utf-8")) > 32 * 1024:
return None
return {"instance_id": instance_id, "state": state}
if message_type == UI_CLOSE and set(payload) == {"instance_id"}:
return {"instance_id": instance_id}
return None
def validate_tool_call_payload(payload: dict[str, Any], *, now: datetime | None = None) -> tuple[dict[str, Any], dict[str, Any]] | None:
"""Return the canonical tool wire payload and its minimal response schema.
The canonicalization intentionally drops unknown keys. In particular it
prevents a model from smuggling HTML, DOM configuration, app identifiers,
or capability arguments through the standard-component path.
"""
call_id = _identifier(payload.get("call_id"))
tool = payload.get("tool")
title = _bounded_text(payload.get("title", ""), 200)
prompt = _bounded_text(payload.get("prompt", ""), 2_000)
expiry = _timestamp(payload.get("expires_at"))
data = _json_object(payload.get("data"))
if not call_id or tool not in _TOOL_KINDS or title is None or prompt is None or not expiry or not data:
return None
reference = now or datetime.now(timezone.utc)
if expiry[1] <= reference.timestamp():
return None
canonical: dict[str, Any] = {
"call_id": call_id,
"tool": tool,
"title": title,
"prompt": prompt,
"expires_at": expiry[0],
}
if tool == "choice":
group = _json_object(data.get("action_group"))
actions = group.get("actions") if group else None
mode = group.get("mode") if group else None
if mode not in _ACTION_MODES or not isinstance(actions, list) or not 1 <= len(actions) <= 12:
return None
seen: set[str] = set()
canonical_actions: list[dict[str, str]] = []
for action in actions:
item = _json_object(action)
action_id = _identifier(item.get("id")) if item else None
label = _bounded_text(item.get("label"), 200, allow_empty=False) if item else None
description = _bounded_text(item.get("description", ""), 500) if item else None
if not action_id or not label or description is None or action_id in seen:
return None
seen.add(action_id)
entry = {"id": action_id, "label": label}
if description:
entry["description"] = description
canonical_actions.append(entry)
canonical["data"] = {"action_group": {"mode": mode, "actions": canonical_actions}}
return canonical, {"tool": tool, "mode": mode, "action_ids": sorted(seen)}
if tool == "confirm":
confirm = _json_object(data.get("confirm"))
if not confirm:
return None
approve = _bounded_text(confirm.get("approve_label", "确认"), 80, allow_empty=False)
cancel = _bounded_text(confirm.get("cancel_label", "取消"), 80, allow_empty=False)
if not approve or not cancel:
return None
canonical["data"] = {"confirm": {"approve_label": approve, "cancel_label": cancel}}
return canonical, {"tool": tool}
form = _json_object(data.get("form"))
fields = form.get("fields") if form else None
if not isinstance(fields, list) or not 1 <= len(fields) <= 8:
return None
seen_fields: set[str] = set()
canonical_fields: list[dict[str, Any]] = []
schema_fields: dict[str, dict[str, Any]] = {}
for field in fields:
item = _json_object(field)
field_id = _identifier(item.get("id")) if item else None
label = _bounded_text(item.get("label"), 120, allow_empty=False) if item else None
field_type = item.get("type") if item else None
placeholder = _bounded_text(item.get("placeholder", ""), 200) if item else None
pattern = _bounded_text(item.get("pattern", ""), 256) if item else None
minimum = item.get("min_length") if item else None
maximum = item.get("max_length") if item else None
if (not field_id or not label or field_type not in _FIELD_TYPES or placeholder is None or pattern is None
or field_id in seen_fields or (minimum is not None and (not isinstance(minimum, int) or isinstance(minimum, bool) or not 0 <= minimum <= 10_000))
or (maximum is not None and (not isinstance(maximum, int) or isinstance(maximum, bool) or not 0 <= maximum <= 10_000))
or (minimum is not None and maximum is not None and minimum > maximum)):
return None
if pattern:
try:
re.compile(pattern)
except re.error:
return None
seen_fields.add(field_id)
entry: dict[str, Any] = {"id": field_id, "label": label, "type": field_type, "required": item.get("required") is True}
if placeholder:
entry["placeholder"] = placeholder
if minimum is not None:
entry["min_length"] = minimum
if maximum is not None:
entry["max_length"] = maximum
if pattern:
entry["pattern"] = pattern
canonical_fields.append(entry)
# The durable response schema contains only field ids and constraints,
# never labels, prompts, or user-entered values.
schema_fields[field_id] = {"type": field_type, "required": item.get("required") is True, "min_length": minimum, "max_length": maximum}
submit = _bounded_text(form.get("submit_label", "提交"), 80, allow_empty=False) if form else None
cancel = _bounded_text(form.get("cancel_label", "取消"), 80, allow_empty=False) if form else None
if not submit or not cancel:
return None
canonical["data"] = {"form": {"fields": canonical_fields, "submit_label": submit, "cancel_label": cancel}}
return canonical, {"tool": tool, "fields": schema_fields}
def parse_agent_tool_call(raw: str) -> tuple[dict[str, Any], dict[str, Any], float] | None:
"""Return a validated model tool call plus durable metadata, if present."""
try:
candidate = json.loads(raw)
except (TypeError, ValueError):
return None
if not isinstance(candidate, dict) or candidate.get("v") != PROTOCOL_VERSION or candidate.get("type") != TOOL_CALL:
return None
payload = _json_object(candidate.get("payload"))
validated = validate_tool_call_payload(payload) if payload else None
if not validated:
return None
expiry = _timestamp(validated[0]["expires_at"])
assert expiry is not None
return validated[0], validated[1], expiry[1]
def parse_agent_app_call(raw: str, inventory: dict[str, Any] | None) -> tuple[dict[str, Any], float] | None:
"""Return an inventory-bound, canonical Agent app.call before sending it."""
try:
candidate = json.loads(raw)
except (TypeError, ValueError):
return None
if not isinstance(candidate, dict) or candidate.get("v") != PROTOCOL_VERSION or candidate.get("type") != APP_CALL:
return None
payload = _json_object(candidate.get("payload"))
return validate_app_call_payload(payload, inventory) if payload else None
def parse_agent_tool_invoke(raw: str, inventory: dict[str, Any] | None) -> dict[str, Any] | None:
"""Return an inventory-bound, canonical Agent tool.invoke before sending it."""
try:
candidate = json.loads(raw)
except (TypeError, ValueError):
return None
if not isinstance(candidate, dict) or candidate.get("v") != PROTOCOL_VERSION or candidate.get("type") != TOOL_INVOKE:
return None
payload = _json_object(candidate.get("payload"))
return validate_tool_invoke_payload(payload, inventory) if payload else None
def parse_tool_response(raw_payload: str) -> ToolResponse | None:
"""Decode only a valid M1 client tool response; all other input is None."""
try:
value = json.loads(raw_payload)
except (TypeError, ValueError):
return None
if not isinstance(value, dict) or value.get("v") != PROTOCOL_VERSION:
return None
payload = _json_object(value.get("payload"))
call_id = _identifier(payload.get("call_id")) if payload else None
if not call_id:
return None
if value.get("type") == TOOL_RESULT:
status = payload.get("status")
result = _json_object(payload.get("result"))
if status not in _RESULT_STATUSES or result is None:
return None
return ToolResponse(kind="result", call_id=call_id, status=status, result=result)
if value.get("type") == TOOL_CANCEL:
reason = _bounded_text(payload.get("reason", ""), 500)
return ToolResponse(kind="cancel", call_id=call_id, reason=reason) if reason is not None else None
return None
def validate_tool_response(response: ToolResponse, schema: dict[str, Any]) -> dict[str, Any] | None:
"""Build the compact model-facing result from a call's stored schema."""
tool = schema.get("tool")
if response.kind == "cancel":
return {"call_id": response.call_id, "outcome": "cancelled", "reason": response.reason or ""}
if response.status != "completed" or not isinstance(response.result, dict):
# A client may report an expiry/failure, but that must not be reified as
# arbitrary structured input to the model.
return {"call_id": response.call_id, "outcome": response.status or "failed"}
if tool == "choice":
selected = response.result.get("action_ids")
allowed = set(schema.get("action_ids", []))
mode = schema.get("mode")
# The Host emits the canonical single-choice projection as one
# `action_id`; normalize it to the Adapter's internal list shape.
if selected is None and mode == "single-choice":
action_id = response.result.get("action_id")
selected = [action_id] if isinstance(action_id, str) else None
if not isinstance(selected, list) or not selected or any(not isinstance(value, str) or value not in allowed for value in selected):
return None
if len(set(selected)) != len(selected) or (mode in {"single-choice", "button"} and len(selected) != 1):
return None
return {"call_id": response.call_id, "outcome": "completed", "action_ids": selected}
if tool == "confirm":
if response.result.get("confirmed") is not True or set(response.result) != {"confirmed"}:
return None
return {"call_id": response.call_id, "outcome": "confirmed"}
if tool == "input":
values = response.result.get("values")
fields = schema.get("fields")
if not isinstance(fields, dict):
return None
# The Host may emit a single-field input result as a compact `text`
# value instead of the full `values.{field_id}` map. Normalize it to
# the schema-bound internal shape before validation.
if values is None and len(fields) == 1:
text_value = response.result.get("text")
if isinstance(text_value, str):
field_id = next(iter(fields))
values = {field_id: text_value}
if not isinstance(values, dict) or set(values) != set(fields):
return None
clean_values: dict[str, str] = {}
for field_id, definition in fields.items():
value = values.get(field_id)
if not isinstance(value, str) or len(value) > 10_000:
return None
minimum = definition.get("min_length")
maximum = definition.get("max_length")
if (definition.get("required") and not value.strip()) or (minimum is not None and len(value) < minimum) or (maximum is not None and len(value) > maximum):
return None
clean_values[field_id] = value
return {"call_id": response.call_id, "outcome": "completed", "values": clean_values}
return None
def message_id() -> str:
return f"msg_{uuid.uuid4().hex}"
def envelope(
message_type: str,
payload: dict[str, Any],
*,
conversation_id: str,
sender_id: str,
sender_kind: str = "agent",
target_id: str | None = None,
target_kind: str = "human",
) -> dict[str, Any]:
result: dict[str, Any] = {
"v": PROTOCOL_VERSION,
"id": message_id(),
"type": message_type,
"conversation_id": conversation_id,
"sender": {"kind": sender_kind, "id": sender_id},
"timestamp": datetime.now(timezone.utc).isoformat(),
"payload": payload,
}
if target_id:
result["target"] = {"kind": target_kind, "id": target_id}
return result
def encode(value: dict[str, Any]) -> str:
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
def text_message(
text: str, *, conversation_id: str, sender_id: str, target_id: str | None = None
) -> str:
return encode(
envelope(
TEXT,
{"text": text},
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def status_message(
status: str,
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
detail: str | None = None,
execution_id: str | None = None,
) -> str:
payload: dict[str, Any] = {"status": status}
if detail:
payload["detail"] = detail
if execution_id and _identifier(execution_id):
payload["execution_id"] = execution_id
return encode(
envelope(
STATUS,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def execution_message(
execution_id: str,
status: str,
steps: list[dict[str, str]],
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
) -> str:
"""Emit an adapter-owned, display-only operation trace.
Callers must already have reduced tool ledger data to the small public
schema. This helper intentionally has no command/result/metadata fields.
"""
if not _identifier(execution_id) or status not in {"running", "completed", "failed"}:
raise ValueError("invalid execution trace")
canonical: list[dict[str, str]] = []
seen: set[str] = set()
for step in steps[:12]:
step_id = _identifier(step.get("id"))
title = _bounded_text(step.get("title"), 160, allow_empty=False)
step_status = step.get("status")
if not step_id or not title or step_status not in {"running", "completed", "failed"} or step_id in seen:
continue
seen.add(step_id)
canonical.append({"id": step_id, "title": title, "status": step_status})
return encode(envelope(EXECUTION, {"execution_id": execution_id, "status": status, "steps": canonical}, conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
def error_message(
code: str,
message: str,
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
) -> str:
return encode(
envelope(
ERROR,
{"code": code, "message": message},
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def tool_result_message(
call_id: str,
status: str,
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
error: str | None = None,
) -> str:
"""Emit an adapter-authoritative terminal state for one standard tool."""
payload: dict[str, Any] = {"call_id": call_id, "status": status, "result": {}}
if error:
payload["error"] = error
return encode(
envelope(
TOOL_RESULT,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def ui_open_message(
instance_id: str,
app: dict[str, Any],
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
state: dict[str, Any] | None = None,
) -> str:
"""Create a sandboxed UI Surface request.
``app`` is deliberately transport-only here. The receiving LineUp Client
owns the sandbox, validates its size/version, and never grants the UI
bundle access to client credentials or native app capabilities.
"""
payload: dict[str, Any] = {"instance_id": instance_id, "app": app}
if state is not None:
payload["state"] = state
return encode(
envelope(
UI_OPEN,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def ui_event_message(
instance_id: str,
event: str,
data: dict[str, Any] | None,
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
) -> str:
"""Return a user interaction from a UI Surface to its owning Agent."""
return encode(
envelope(
UI_EVENT,
{"instance_id": instance_id, "event": event, "data": data or {}},
conversation_id=conversation_id,
sender_id=sender_id,
sender_kind="human",
target_id=target_id,
target_kind="agent",
)
)
def app_call_message(
call_id: str,
capability: str,
arguments: dict[str, Any],
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
reason: str | None = None,
expires_at: str | None = None,
) -> str:
"""Request one registered host-app capability.
A client must still verify its local registry and obtain the user/system
authorization required by that capability before it executes anything.
"""
payload: dict[str, Any] = {
"call_id": call_id,
"capability": capability,
"arguments": arguments,
}
if reason:
payload["reason"] = reason
if expires_at:
payload["expires_at"] = expires_at
return encode(
envelope(
APP_CALL,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
def app_result_message(
call_id: str,
status: str,
*,
conversation_id: str,
sender_id: str,
target_id: str | None = None,
result: dict[str, Any] | None = None,
error: str | None = None,
) -> str:
"""Return the definitive result of an app capability call."""
payload: dict[str, Any] = {"call_id": call_id, "status": status, "result": result or {}}
if error:
payload["error"] = error
return encode(
envelope(
APP_RESULT,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
sender_kind="human",
target_id=target_id,
target_kind="agent",
)
)
def agent_output_message(
raw: str, *, conversation_id: str, sender_id: str, target_id: str | None = None, execution_id: str | None = None,
inventory: dict[str, Any] | None = None,
) -> str:
"""Safely promote a model-produced LineUp UI envelope, or wrap it as text.
Hermes normally returns Markdown. When it deliberately returns one JSON
object with an allowlisted LineUp type, the adapter rebuilds the outer
envelope itself so a model cannot spoof actor metadata or emit a
client-only result type.
"""
try:
candidate = json.loads(raw)
except (TypeError, ValueError):
candidate = None
if not isinstance(candidate, dict):
payload: dict[str, Any] = {"text": str(raw)}
if execution_id and _identifier(execution_id):
payload["execution_id"] = execution_id
return encode(envelope(TEXT, payload, conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
message_type = candidate.get("type")
payload = candidate.get("payload")
if candidate.get("v") == PROTOCOL_VERSION and message_type == APP_CALL and isinstance(payload, dict):
validated_app_call = validate_app_call_payload(payload, inventory)
if not validated_app_call:
return text_message(str(raw), conversation_id=conversation_id, sender_id=sender_id, target_id=target_id)
return encode(envelope(APP_CALL, validated_app_call[0], conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
if candidate.get("v") == PROTOCOL_VERSION and message_type == TOOL_INVOKE and isinstance(payload, dict):
validated_tool_invoke = validate_tool_invoke_payload(payload, inventory)
if not validated_tool_invoke:
return text_message(str(raw), conversation_id=conversation_id, sender_id=sender_id, target_id=target_id)
return encode(envelope(TOOL_INVOKE, validated_tool_invoke, conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
if candidate.get("v") == PROTOCOL_VERSION and message_type in AGENT_OUTPUT_TYPES and isinstance(payload, dict):
if message_type == TEXT:
# The execution association is adapter-owned. Rebuild even a
# model-shaped text envelope so the model cannot drop, spoof, or
# add unrelated metadata to this UI-only relationship.
rendered = payload.get("markdown") if isinstance(payload.get("markdown"), str) else payload.get("text")
if not isinstance(rendered, str):
rendered = str(raw)
canonical_text: dict[str, Any] = {"text": rendered}
if execution_id and _identifier(execution_id):
canonical_text["execution_id"] = execution_id
return encode(envelope(TEXT, canonical_text, conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
if message_type == TOOL_CALL:
validated = validate_tool_call_payload(payload)
if not validated:
return text_message(str(raw), conversation_id=conversation_id, sender_id=sender_id, target_id=target_id)
payload = validated[0]
if message_type in {UI_OPEN, UI_PATCH, UI_CLOSE}:
payload = validate_surface_payload(message_type, payload)
if not payload:
return text_message(str(raw), conversation_id=conversation_id, sender_id=sender_id, target_id=target_id)
if message_type == ARTIFACT_OFFER:
payload = validate_artifact_offer_payload(payload)
if not payload:
return text_message(str(raw), conversation_id=conversation_id, sender_id=sender_id, target_id=target_id)
return encode(
envelope(
message_type,
payload,
conversation_id=conversation_id,
sender_id=sender_id,
target_id=target_id,
)
)
payload = {"text": str(raw)}
if execution_id and _identifier(execution_id):
payload["execution_id"] = execution_id
return encode(envelope(TEXT, payload, conversation_id=conversation_id, sender_id=sender_id, target_id=target_id))
def extract_user_input(raw_payload: str) -> tuple[str, str | None]:
"""Return LLM-facing text and an optional client-provided conversation ID.
The legacy bridge is intentionally input-only. All adapter output uses the
versioned envelope above.
"""
try:
value = json.loads(raw_payload)
except (TypeError, ValueError):
return str(raw_payload), None
if not isinstance(value, dict):
return str(raw_payload), None
conversation_id = value.get("conversation_id")
if not isinstance(conversation_id, str) or not conversation_id.strip():
conversation_id = None
message_type = value.get("type")
payload = value.get("payload")
if message_type == TEXT and isinstance(payload, dict) and isinstance(payload.get("text"), str):
return payload["text"], conversation_id
if message_type == TOOL_RESULT and isinstance(payload, dict):
# ``LineUpAdapter`` consumes this through ``parse_tool_response`` and
# its durable call ledger before it can reach Hermes. Keeping this
# compatibility text non-structured prevents direct callers from
# passing an unvalidated UI-shaped object to the model.
return "LineUp standard interaction response received.", conversation_id
if message_type == UI_EVENT and isinstance(payload, dict):
return "LineUp UI surface event:\n" + json.dumps(payload, ensure_ascii=False), conversation_id
if message_type == APP_RESULT and isinstance(payload, dict):
# Core consumes a separately validated and reduced app result. Never
# expose a raw client payload through this legacy compatibility path.
return "LineUp app capability result received.", conversation_id
if message_type == TOOL_CANCEL and isinstance(payload, dict):
return "LineUp standard interaction cancellation received.", conversation_id
if message_type == "choice.response":
return str(value.get("value", "")), conversation_id
return str(raw_payload), conversation_id
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env bash
# Manage the real Hermes-backed LineUp Adapter.
set -euo pipefail
DIR="$(cd "$(dirname "$0")" && pwd)"
LOG="$DIR/hermes-adapter.log"
PIDFILE="$DIR/hermes-adapter.pid"
PYTHON_BIN="${LINEUP_PYTHON_BIN:-python3}"
RUNTIME_ENV="$DIR/.env"
# Keep deployment-only credentials out of the plugin manifest and repository.
if [[ -f "$RUNTIME_ENV" ]]; then
set -a
# shellcheck disable=SC1090
source "$RUNTIME_ENV"
set +a
fi
is_running() {
[[ -f "$PIDFILE" ]] && kill -0 "$(<"$PIDFILE")" 2>/dev/null
}
start() {
if is_running; then
echo "Hermes Adapter 已在运行 PID=$(<"$PIDFILE")"
return
fi
cd "$DIR/.."
# Detach from the invoking terminal/session. This keeps the Adapter alive
# when installed or started by an automation runner that tears down its
# process group after the command returns.
nohup setsid "$PYTHON_BIN" -m lineup.core >> "$LOG" 2>&1 < /dev/null &
echo "$!" > "$PIDFILE"
echo "Hermes Adapter 已启动 PID=$(<"$PIDFILE")"
}
stop() {
if is_running; then
kill "$(<"$PIDFILE")"
echo "Hermes Adapter 已停止"
fi
rm -f "$PIDFILE"
}
case "${1:-start}" in
start) start ;;
stop) stop ;;
restart) stop; start ;;
status) is_running && echo "Hermes Adapter 运行中 PID=$(<"$PIDFILE")" || echo "Hermes Adapter 未运行" ;;
check) cd "$DIR/.."; exec "$PYTHON_BIN" -m lineup.core --check ;;
*) echo "用法: $0 {start|stop|restart|status|check}"; exit 2 ;;
esac
+27
View File
@@ -0,0 +1,27 @@
#!/bin/bash
# Install LineUp Hermes plugin to Hermes plugins directory
set -euo pipefail
PLUGIN_DIR="${HOME}/.hermes/plugins/lineup"
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
PLUGIN_SOURCE="$(cd "${SCRIPT_DIR}/.." && pwd)"
echo "Installing LineUp plugin to ${PLUGIN_DIR}..."
mkdir -p "${PLUGIN_DIR}"
rsync -av --delete \
--exclude='.venv' --exclude='.git' --exclude='*.egg-info' \
--exclude='state.sqlite3*' --exclude='*.log' --exclude='*.pid' --exclude='__pycache__' \
"${PLUGIN_SOURCE}/" "${PLUGIN_DIR}/"
echo "Done. Enable the plugin with: hermes plugins enable lineup"
echo ""
echo "Run the adapter separately (recommended):"
echo " ${PLUGIN_DIR}/run-hermes-adapter.sh start"
echo ""
echo "Required runtime environment:"
echo " LINEUP_APPSERVER_URL=http://127.0.0.1:8090"
echo " LINEUP_AGENT_UID=agent_hermes_main"
echo " LINEUP_AGENT_SHARED_SECRET=<configured server secret>"
echo ""
echo "Set LINEUP_ADAPTER_AUTOSTART=1 only when Hermes itself is a long-running process."
+357
View File
@@ -0,0 +1,357 @@
"""Durable inbox/outbox state for the LineUp Hermes Adapter (stdlib only)."""
from __future__ import annotations
import json
import sqlite3
import time
from pathlib import Path
from typing import Any
class StateStore:
def __init__(self, path: str | Path):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self.db = sqlite3.connect(self.path, check_same_thread=False)
self.db.row_factory = sqlite3.Row
self.db.execute("PRAGMA journal_mode=WAL")
self.db.executescript(
"""
CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS inbox (
message_id TEXT PRIMARY KEY,
message_seq INTEGER NOT NULL,
from_uid TEXT NOT NULL,
payload TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'pending',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
error TEXT
);
CREATE TABLE IF NOT EXISTS outbox (
id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
created_at INTEGER NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
CREATE TABLE IF NOT EXISTS tool_calls (
conversation_id TEXT NOT NULL,
call_id TEXT NOT NULL,
tool TEXT NOT NULL,
expires_at REAL NOT NULL,
response_schema TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'emitted',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (conversation_id, call_id)
);
CREATE TABLE IF NOT EXISTS tool_audit (
id INTEGER PRIMARY KEY AUTOINCREMENT,
conversation_id TEXT NOT NULL,
call_id TEXT NOT NULL,
event TEXT NOT NULL,
disposition TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS tool_audit_conversation_call
ON tool_audit(conversation_id, call_id, id);
CREATE TABLE IF NOT EXISTS app_calls (
conversation_id TEXT NOT NULL,
call_id TEXT NOT NULL,
capability TEXT NOT NULL,
expires_at REAL NOT NULL,
state TEXT NOT NULL DEFAULT 'emitted',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (conversation_id, call_id)
);
CREATE TABLE IF NOT EXISTS app_audit (
id INTEGER PRIMARY KEY AUTOINCREMENT,
conversation_id TEXT NOT NULL,
call_id TEXT NOT NULL,
event TEXT NOT NULL,
disposition TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS hermes_interactions (
conversation_id TEXT NOT NULL,
call_id TEXT NOT NULL,
kind TEXT NOT NULL,
session_id TEXT NOT NULL,
acp_request_id INTEGER NOT NULL,
option_map TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'pending',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (conversation_id, call_id)
);
"""
)
self.db.commit()
def close(self) -> None:
self.db.close()
def cursor(self) -> int:
row = self.db.execute("SELECT value FROM metadata WHERE key='cursor'").fetchone()
return int(row["value"]) if row else 0
def set_cursor(self, value: int) -> None:
self.db.execute(
"INSERT INTO metadata(key, value) VALUES('cursor', ?) "
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
(str(value),),
)
self.db.commit()
def is_bootstrapped(self) -> bool:
row = self.db.execute("SELECT value FROM metadata WHERE key='bootstrapped'").fetchone()
return bool(row and row["value"] == "1")
def set_bootstrapped(self) -> None:
self.db.execute(
"INSERT INTO metadata(key, value) VALUES('bootstrapped', '1') "
"ON CONFLICT(key) DO UPDATE SET value='1'"
)
self.db.commit()
def claim_inbox(self, message: dict[str, Any]) -> bool:
now = int(time.time())
message_id = str(message.get("message_id") or f"seq:{message.get('message_seq')}")
cur = self.db.execute(
"INSERT OR IGNORE INTO inbox(message_id, message_seq, from_uid, payload, state, created_at, updated_at) "
"VALUES (?, ?, ?, ?, 'pending', ?, ?)",
(
message_id,
int(message.get("message_seq") or 0),
str(message.get("from_uid") or ""),
str(message.get("payload") or ""),
now,
now,
),
)
self.db.commit()
return cur.rowcount == 1
def pending(self) -> list[sqlite3.Row]:
return self.db.execute(
"SELECT * FROM inbox WHERE state IN ('pending', 'running') ORDER BY message_seq ASC"
).fetchall()
def recover_interrupted_turns(self) -> None:
"""A subprocess cannot survive an adapter restart.
Returning its inbox rows to ``pending`` gives the user message one
deterministic retry instead of leaving it permanently stuck in the
old worker's ``running`` state.
"""
self.db.execute(
"UPDATE inbox SET state='pending', updated_at=?, error='adapter restarted before Hermes completed' "
"WHERE state='running'",
(int(time.time()),),
)
self.db.commit()
def recover_pending_hermes_interactions(self) -> None:
"""Expire adapter-owned interactive bridges that cannot survive restart."""
now = int(time.time())
self.db.execute(
"UPDATE tool_calls SET state='expired', updated_at=? "
"WHERE state='emitted' AND EXISTS ("
" SELECT 1 FROM hermes_interactions hi "
" WHERE hi.conversation_id=tool_calls.conversation_id AND hi.call_id=tool_calls.call_id AND hi.state='pending'"
")",
(now,),
)
self.db.execute(
"UPDATE hermes_interactions SET state='aborted', updated_at=? WHERE state='pending'",
(now,),
)
self.db.commit()
def set_inbox_state(self, message_id: str, state: str, error: str | None = None) -> None:
self.db.execute(
"UPDATE inbox SET state=?, updated_at=?, error=? WHERE message_id=?",
(state, int(time.time()), error, message_id),
)
self.db.commit()
def register_tool_call(
self, conversation_id: str, call_id: str, tool: str, expires_at: float, response_schema: dict[str, Any]
) -> str:
"""Durably reserve a call before it can be sent to the client."""
now = int(time.time())
cur = self.db.execute(
"INSERT OR IGNORE INTO tool_calls(conversation_id, call_id, tool, expires_at, response_schema, state, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, 'emitted', ?, ?)",
(conversation_id, call_id, tool, expires_at, json.dumps(response_schema, ensure_ascii=False, separators=(",", ":")), now, now),
)
disposition = "accepted" if cur.rowcount == 1 else "duplicate"
self._audit_tool(conversation_id, call_id, "call", disposition, now)
self.db.commit()
return disposition
def tool_call(self, conversation_id: str, call_id: str) -> sqlite3.Row | None:
return self.db.execute(
"SELECT * FROM tool_calls WHERE conversation_id=? AND call_id=?", (conversation_id, call_id)
).fetchone()
def register_hermes_interaction(
self,
conversation_id: str,
call_id: str,
*,
kind: str,
session_id: str,
acp_request_id: int,
option_map: dict[str, Any],
) -> str:
now = int(time.time())
cur = self.db.execute(
"INSERT OR IGNORE INTO hermes_interactions("
"conversation_id, call_id, kind, session_id, acp_request_id, option_map, state, created_at, updated_at"
") VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)",
(
conversation_id,
call_id,
kind,
session_id,
acp_request_id,
json.dumps(option_map, ensure_ascii=False, separators=(",", ":")),
now,
now,
),
)
self.db.commit()
return "accepted" if cur.rowcount == 1 else "duplicate"
def hermes_interaction(self, conversation_id: str, call_id: str) -> sqlite3.Row | None:
return self.db.execute(
"SELECT * FROM hermes_interactions WHERE conversation_id=? AND call_id=?",
(conversation_id, call_id),
).fetchone()
def complete_hermes_interaction(self, conversation_id: str, call_id: str, state: str) -> None:
self.db.execute(
"UPDATE hermes_interactions SET state=?, updated_at=? WHERE conversation_id=? AND call_id=?",
(state, int(time.time()), conversation_id, call_id),
)
self.db.commit()
def consume_tool_response(self, conversation_id: str, call_id: str, event: str) -> str:
"""Atomically claim one client result/cancellation without storing it.
The result itself can contain private form values, so it is forwarded to
the model only in memory after schema validation and never enters this
ledger or its audit trail.
"""
now = int(time.time())
row = self.tool_call(conversation_id, call_id)
if not row:
self._audit_tool(conversation_id, call_id, event, "rejected", now)
self.db.commit()
return "rejected"
if float(row["expires_at"]) <= time.time():
self.db.execute(
"UPDATE tool_calls SET state='expired', updated_at=? WHERE conversation_id=? AND call_id=? AND state='emitted'",
(now, conversation_id, call_id),
)
self._audit_tool(conversation_id, call_id, event, "expired", now)
self.db.commit()
return "expired"
if row["state"] != "emitted":
self._audit_tool(conversation_id, call_id, event, "duplicate", now)
self.db.commit()
return "duplicate"
self.db.execute(
"UPDATE tool_calls SET state=?, updated_at=? WHERE conversation_id=? AND call_id=? AND state='emitted'",
("completed" if event == "result" else "cancelled", now, conversation_id, call_id),
)
self._audit_tool(conversation_id, call_id, event, "accepted", now)
self.db.commit()
return "accepted"
def tool_audit(self, conversation_id: str, call_id: str) -> list[sqlite3.Row]:
return self.db.execute(
"SELECT event, disposition FROM tool_audit WHERE conversation_id=? AND call_id=? ORDER BY id ASC",
(conversation_id, call_id),
).fetchall()
def record_tool_audit(self, conversation_id: str, call_id: str, event: str, disposition: str) -> None:
self._audit_tool(conversation_id, call_id, event, disposition, int(time.time()))
self.db.commit()
def register_app_call(self, conversation_id: str, call_id: str, capability: str, expires_at: float) -> str:
"""Durably reserve a capability call before it can reach the host."""
now = int(time.time())
cur = self.db.execute(
"INSERT OR IGNORE INTO app_calls(conversation_id, call_id, capability, expires_at, state, created_at, updated_at) "
"VALUES (?, ?, ?, ?, 'emitted', ?, ?)",
(conversation_id, call_id, capability, expires_at, now, now),
)
disposition = "accepted" if cur.rowcount == 1 else "duplicate"
self._audit_app(conversation_id, call_id, "call", disposition, now)
self.db.commit()
return disposition
def consume_app_result(self, conversation_id: str, call_id: str, status: str) -> str:
"""Atomically accept one terminal host result without storing payload data."""
now = int(time.time())
row = self.db.execute("SELECT * FROM app_calls WHERE conversation_id=? AND call_id=?", (conversation_id, call_id)).fetchone()
if not row:
self._audit_app(conversation_id, call_id, "result", "rejected", now)
self.db.commit()
return "rejected"
if float(row["expires_at"]) <= time.time() and row["state"] == "emitted":
self.db.execute("UPDATE app_calls SET state='expired', updated_at=? WHERE conversation_id=? AND call_id=? AND state='emitted'", (now, conversation_id, call_id))
self._audit_app(conversation_id, call_id, "result", "expired", now)
self.db.commit()
return "expired"
if row["state"] != "emitted":
self._audit_app(conversation_id, call_id, "result", "duplicate", now)
self.db.commit()
return "duplicate"
self.db.execute("UPDATE app_calls SET state=?, updated_at=? WHERE conversation_id=? AND call_id=? AND state='emitted'", (status, now, conversation_id, call_id))
self._audit_app(conversation_id, call_id, "result", "accepted", now)
self.db.commit()
return "accepted"
def _audit_app(self, conversation_id: str, call_id: str, event: str, disposition: str, now: int) -> None:
self.db.execute("INSERT INTO app_audit(conversation_id, call_id, event, disposition, created_at) VALUES (?, ?, ?, ?, ?)", (conversation_id, call_id, event, disposition, now))
def _audit_tool(self, conversation_id: str, call_id: str, event: str, disposition: str, now: int) -> None:
self.db.execute(
"INSERT INTO tool_audit(conversation_id, call_id, event, disposition, created_at) VALUES (?, ?, ?, ?, ?)",
(conversation_id, call_id, event, disposition, now),
)
def enqueue_outbox(self, payload: str) -> str:
item_id = f"out_{time.time_ns()}"
self.db.execute(
"INSERT INTO outbox(id, payload, created_at) VALUES (?, ?, ?)",
(item_id, payload, int(time.time())),
)
self.db.commit()
return item_id
def outbox_items(self) -> list[sqlite3.Row]:
return self.db.execute("SELECT * FROM outbox ORDER BY created_at ASC").fetchall()
def mark_outbox_sent(self, item_id: str) -> None:
self.db.execute("DELETE FROM outbox WHERE id=?", (item_id,))
self.db.commit()
def mark_outbox_failed(self, item_id: str, error: str) -> None:
self.db.execute(
"UPDATE outbox SET attempts=attempts+1, last_error=? WHERE id=?", (error[:1000], item_id)
)
self.db.commit()
def status_summary(self) -> str:
cursor = self.cursor()
pending = self.db.execute("SELECT count(*) AS n FROM inbox WHERE state != 'done'").fetchone()["n"]
outbox = self.db.execute("SELECT count(*) AS n FROM outbox").fetchone()["n"]
active_tools = self.db.execute("SELECT count(*) AS n FROM tool_calls WHERE state='emitted'").fetchone()["n"]
return json.dumps({"cursor": cursor, "pending_inbox": pending, "pending_outbox": outbox, "active_tools": active_tools}, ensure_ascii=False)
+98
View File
@@ -0,0 +1,98 @@
from __future__ import annotations
import sys
import threading
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from lineup.acp_client import HermesACPClient, public_tool_update
class ACPProjectionTest(unittest.TestCase):
def test_projects_real_hermes_lifecycle_without_exposing_arguments_or_titles(self) -> None:
started = public_tool_update({
"sessionUpdate": "tool_call", "toolCallId": "tool_001", "kind": "fetch",
"title": "curl http://private-host/health?token=secret",
"content": [{"type": "content", "text": "private token"}],
})
completed = public_tool_update({
"sessionUpdate": "tool_call_update", "toolCallId": "tool_001", "status": "completed",
"content": [{"type": "content", "text": "private output"}],
}, {"tool_001": "读取公开网页内容"})
self.assertEqual(("tool_001", "读取公开网页内容", "running"), (started.tool_id, started.title, started.status))
self.assertEqual(("tool_001", "读取公开网页内容", "completed"), (completed.tool_id, completed.title, completed.status))
def test_projects_real_kind_categories_and_terminal_failure(self) -> None:
labels = {
"read": "读取相关配置",
"search": "检索相关文件",
"fetch": "读取公开网页内容",
"execute": "执行已批准的系统检查",
}
for index, (tool_kind, label) in enumerate(labels.items(), start=1):
tool_id = f"tool_{index:03d}"
started = public_tool_update({"sessionUpdate": "tool_call", "toolCallId": tool_id, "kind": tool_kind})
failed = public_tool_update({"sessionUpdate": "tool_call_update", "toolCallId": tool_id, "status": "failed"}, {tool_id: label})
self.assertEqual((tool_id, label, "running"), (started.tool_id, started.title, started.status))
self.assertEqual((tool_id, label, "failed"), (failed.tool_id, failed.title, failed.status))
def test_rejects_thoughts_unknown_tools_and_unsafe_ids(self) -> None:
self.assertIsNone(public_tool_update({"sessionUpdate": "agent_thought_chunk", "content": {"text": "secret"}}))
self.assertIsNone(public_tool_update({"sessionUpdate": "tool_call", "toolCallId": "1bad", "kind": "execute"}))
self.assertIsNone(public_tool_update({"sessionUpdate": "tool_call", "toolCallId": "tool_001", "kind": "edit"}))
self.assertIsNone(public_tool_update({"sessionUpdate": "tool_call_update", "toolCallId": "tool_001", "status": "completed"}))
def test_projects_permission_request_and_selected_response(self) -> None:
seen = []
sent = []
client = HermesACPClient(
"hermes",
".",
on_update=lambda params: None,
on_request=lambda request: seen.append(request),
on_closed=lambda error: None,
)
client._send = lambda message: sent.append(message) # type: ignore[method-assign]
message = {
"jsonrpc": "2.0",
"id": 7,
"method": "session/request_permission",
"params": {
"sessionId": "session-1",
"toolCall": {
"toolCallId": "perm-check-1",
"title": "dangerous command: rm -rf /tmp/demo",
"content": [{"content": {"type": "text", "text": "dangerous command\n$ rm -rf /tmp/demo"}}],
},
"options": [
{"optionId": "allow_once", "kind": "allow_once", "name": "Allow once"},
{"optionId": "allow_session", "kind": "allow_always", "name": "Allow for session"},
{"optionId": "deny", "kind": "reject_once", "name": "Deny"},
],
},
}
worker = threading.Thread(target=client._handle_server_request, args=(message,))
worker.start()
for _ in range(50):
if seen:
break
worker.join(0.01)
self.assertTrue(seen)
request = seen[0]
self.assertEqual("exec_approval", request.kind)
self.assertEqual("session-1", request.session_id)
self.assertEqual(("action_01", "allow_once"), (request.options[0].action_id, request.options[0].option_id))
self.assertIn("rm -rf /tmp/demo", request.prompt)
self.assertTrue(client.respond_permission(request.request_id, "allow_session"))
worker.join(1.0)
self.assertEqual({
"jsonrpc": "2.0",
"id": 7,
"result": {"outcome": {"optionId": "allow_session", "outcome": "selected"}},
}, sent[0])
+406
View File
@@ -0,0 +1,406 @@
from __future__ import annotations
import sys
import tempfile
import unittest
import json
import time
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from lineup.acp_client import ACPInteractiveOption, ACPInteractiveRequest
from lineup.core import AdapterConfig, LineUpAdapter, RunningTurn
from lineup.state import StateStore
class CoreIsolationTest(unittest.TestCase):
@staticmethod
def _config(temp_dir: str) -> AdapterConfig:
return AdapterConfig(
appserver_url="http://127.0.0.1:8090", agent_uid="agent", channel_id="channel-a", channel_type=2,
poll_interval=2, request_timeout=1, hermes_timeout=10, hermes_bin="hermes", hermes_cwd=temp_dir,
state_path=str(Path(temp_dir) / "state.sqlite3"), hermes_state_path=str(Path(temp_dir) / "hermes-state.db"),
agent_secret="", ignored_senders=frozenset(), replay_history=False,
)
def test_client_claimed_conversation_cannot_switch_the_hermes_session(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
config = self._config(temp_dir)
adapter = LineUpAdapter(config)
try:
canonical = adapter._conversation_id("user-a", "attacker-selected-conversation")
self.assertEqual("lineup:2:channel-a:user-a", canonical)
self.assertNotEqual(canonical, adapter._conversation_id("user-b", "attacker-selected-conversation"))
self.assertEqual(adapter._session_name(canonical), adapter._session_name(adapter._conversation_id("user-a", None)))
finally:
adapter.close()
def test_acp_tool_event_is_sent_immediately_with_a_safe_public_label(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
class Client:
session_id = "session-1"
def close(self) -> None: pass
running = RunningTurn("message-1", "user", "conv", None, time.monotonic(), "exec_001", "marker", acp=Client())
adapter.running["user"] = running
adapter.acp_events.put(("update", {"sessionId": "session-1", "update": {
"sessionUpdate": "tool_call", "toolCallId": "tool_001", "kind": "fetch",
"title": "curl http://private-host/health?token=secret",
"content": [{"type": "content", "text": "secret"}],
}}, None))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
wire = json.loads(send.call_args.args[0])
self.assertEqual("lineup.v1.agent.execution", wire["type"])
self.assertEqual("读取公开网页内容", wire["payload"]["steps"][0]["title"])
self.assertNotIn("private-host", str(wire))
self.assertNotIn("secret", str(wire))
finally:
adapter.close()
def test_timeout_finishes_its_exact_execution_without_inventing_steps(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
process = __import__("subprocess").Popen(["true"], text=True, stdout=__import__("subprocess").PIPE, stderr=__import__("subprocess").PIPE)
adapter.running["user"] = RunningTurn("message-1", "user", "conv", process, time.monotonic() - 20, "exec_001", "marker")
with patch.object(adapter, "_send_or_queue") as send:
adapter._reap_turns()
trace = json.loads(send.call_args_list[0].args[0])
self.assertEqual("lineup.v1.agent.execution", trace["type"])
self.assertEqual({"execution_id": "exec_001", "status": "failed", "steps": []}, trace["payload"])
finally:
adapter.close()
def test_app_capability_result_has_a_durable_one_shot_ledger(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
try:
expires = time.time() + 60
self.assertEqual("accepted", store.register_app_call("conv", "app-call-1", "app.open_url", expires))
self.assertEqual("duplicate", store.register_app_call("conv", "app-call-1", "app.open_url", expires))
self.assertEqual("accepted", store.consume_app_result("conv", "app-call-1", "rejected"))
self.assertEqual("duplicate", store.consume_app_result("conv", "app-call-1", "rejected"))
self.assertEqual("rejected", store.consume_app_result("conv", "unknown", "completed"))
finally:
store.close()
def test_permission_request_becomes_tool_call_and_user_choice_resolves_acp(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, option_id: str | None) -> bool:
self.responses.append((request_id, option_id))
return True
client = Client()
running = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.running["user"] = running
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="exec_approval",
request_id=11,
session_id="session-1",
title="Hermes 请求权限",
prompt="dangerous command\n$ rm -rf /tmp/demo",
options=(
ACPInteractiveOption(action_id="action_01", option_id="allow_once", label="Allow once"),
ACPInteractiveOption(action_id="action_02", option_id="allow_session", label="Allow for session"),
ACPInteractiveOption(action_id="action_03", option_id="deny", label="Deny"),
),
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
wire = json.loads(send.call_args.args[0])
self.assertEqual("lineup.v1.tool.call", wire["type"])
call_id = wire["payload"]["call_id"]
self.assertEqual("choice", wire["payload"]["tool"])
interaction = adapter.store.hermes_interaction(conversation_id, call_id)
self.assertIsNotNone(interaction)
response_payload = json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": call_id,
"status": "completed",
# This is the exact single-choice shape emitted by the
# Host card interaction.
"result": {"action_id": "action_02"},
},
}, ensure_ascii=False, separators=(",", ":"))
with patch.object(adapter, "_send_or_queue") as send:
adapter._start_turn({
"message_id": "message-2",
"from_uid": "user",
"payload": response_payload,
})
echoed = json.loads(send.call_args.args[0])
self.assertEqual("lineup.v1.tool.result", echoed["type"])
self.assertEqual([(11, "allow_session")], client.responses)
self.assertEqual("resolved", adapter.store.hermes_interaction(conversation_id, call_id)["state"])
finally:
adapter.close()
def test_update_prompt_request_becomes_confirm_and_resolves_yes(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
self.responses.append((request_id, outcome))
return True
client = Client()
adapter.running["user"] = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="update_prompt",
request_id=21,
session_id="session-1",
title="更新提示",
prompt="是否按新的提示继续?",
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
wire = json.loads(send.call_args.args[0])
self.assertEqual("confirm", wire["payload"]["tool"])
call_id = wire["payload"]["call_id"]
response_payload = json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": call_id,
"status": "completed",
"result": {"confirmed": True},
},
}, ensure_ascii=False, separators=(",", ":"))
adapter._start_turn({"message_id": "message-2", "from_uid": "user", "payload": response_payload})
self.assertEqual([(21, "y")], client.responses)
finally:
adapter.close()
def test_clarify_choice_resolves_selected_value(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
self.responses.append((request_id, outcome))
return True
client = Client()
adapter.running["user"] = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="clarify",
request_id=31,
session_id="session-1",
title="需要澄清",
prompt="请选择部署环境。",
options=(
ACPInteractiveOption(action_id="action_01", option_id="staging", label="Staging"),
ACPInteractiveOption(action_id="action_02", option_id="production", label="Production"),
),
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
wire = json.loads(send.call_args.args[0])
call_id = wire["payload"]["call_id"]
response_payload = json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": call_id,
"status": "completed",
"result": {"action_ids": ["action_02"]},
},
}, ensure_ascii=False, separators=(",", ":"))
adapter._start_turn({"message_id": "message-2", "from_uid": "user", "payload": response_payload})
self.assertEqual([(31, "production")], client.responses)
finally:
adapter.close()
def test_slash_confirm_choice_resolves_selected_value(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
self.responses.append((request_id, outcome))
return True
client = Client()
adapter.running["user"] = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="slash_confirm",
request_id=35,
session_id="session-1",
title="确认执行",
prompt="是否继续执行该 slash 命令?",
options=(
ACPInteractiveOption(action_id="action_01", option_id="once", label="仅本次"),
ACPInteractiveOption(action_id="action_02", option_id="always", label="始终允许"),
ACPInteractiveOption(action_id="action_03", option_id="cancel", label="取消"),
),
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
wire = json.loads(send.call_args.args[0])
call_id = wire["payload"]["call_id"]
adapter._start_turn({
"message_id": "message-2",
"from_uid": "user",
"payload": json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": call_id,
"status": "completed",
"result": {"action_ids": ["action_02"]},
},
}, ensure_ascii=False, separators=(",", ":")),
})
self.assertEqual([(35, "always")], client.responses)
finally:
adapter.close()
def test_clarify_other_triggers_followup_input_and_resolves_text(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
self.responses.append((request_id, outcome))
return True
client = Client()
adapter.running["user"] = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="clarify",
request_id=41,
session_id="session-1",
title="需要澄清",
prompt="请选择原因。",
options=(ACPInteractiveOption(action_id="action_01", option_id="known", label="已有选项"),),
allow_other=True,
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
first_wire = json.loads(send.call_args.args[0])
first_call_id = first_wire["payload"]["call_id"]
actions = first_wire["payload"]["data"]["action_group"]["actions"]
other_action_id = next(action["id"] for action in actions if action["label"] == "其他")
with patch.object(adapter, "_send_or_queue") as send:
adapter._start_turn({
"message_id": "message-2",
"from_uid": "user",
"payload": json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": first_call_id,
"status": "completed",
"result": {"action_ids": [other_action_id]},
},
}, ensure_ascii=False, separators=(",", ":")),
})
followup_wire = json.loads(send.call_args.args[0])
self.assertEqual("input", followup_wire["payload"]["tool"])
followup_call_id = followup_wire["payload"]["call_id"]
self.assertEqual([], client.responses)
adapter._start_turn({
"message_id": "message-3",
"from_uid": "user",
"payload": json.dumps({
"v": 1,
"type": "lineup.v1.tool.result",
"payload": {
"call_id": followup_call_id,
"status": "completed",
"result": {"text": "自定义说明"},
},
}, ensure_ascii=False, separators=(",", ":")),
})
self.assertEqual([(41, "自定义说明")], client.responses)
finally:
adapter.close()
def test_clarify_multi_select_is_rejected_without_emitting_host_interaction(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
adapter = LineUpAdapter(self._config(temp_dir))
try:
conversation_id = adapter._conversation_id("user", None)
class Client:
session_id = "session-1"
def __init__(self) -> None:
self.responses: list[tuple[int, str | None]] = []
def close(self) -> None: pass
def respond_interaction(self, request_id: int, outcome: str | None) -> bool:
self.responses.append((request_id, outcome))
return True
client = Client()
adapter.running["user"] = RunningTurn("message-1", "user", conversation_id, None, time.monotonic(), "exec_001", "marker", acp=client)
adapter.acp_clients[conversation_id] = client
request = ACPInteractiveRequest(
kind="clarify",
request_id=51,
session_id="session-1",
title="需要澄清",
prompt="请选择多个选项。",
options=(ACPInteractiveOption(action_id="action_01", option_id="a", label="A"),),
multi_select=True,
)
adapter.acp_events.put(("request", request, client))
with patch.object(adapter, "_send_or_queue") as send:
adapter._flush_acp_events()
send.assert_not_called()
self.assertEqual([(51, None)], client.responses)
finally:
adapter.close()
if __name__ == "__main__":
unittest.main()
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from lineup.execution_trace import extract_public_execution_steps
class ExecutionTraceTest(unittest.TestCase):
def test_projects_only_allowlisted_operations_without_raw_command_data(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "state.db"
database = sqlite3.connect(path)
database.execute("CREATE TABLE messages (id TEXT PRIMARY KEY, session_id TEXT, role TEXT, tool_name TEXT, tool_calls TEXT, content TEXT, timestamp INTEGER)")
database.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?)", ("m1", "session", "user", "", "", "prompt [Internal LineUp turn marker: lineup-turn:exec_001]", 1))
database.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?)", ("m2", "session", "assistant", "terminal", "", "curl http://private-host:8610/health?token=secret qmtbridge", 2))
database.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?)", ("m3", "session", "assistant", "terminal", "", "curl http://private-host/openapi.json", 3))
database.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?)", ("m4", "session", "assistant", "terminal", "", "curl http://private-host/kline", 4))
database.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?)", ("m5", "session", "assistant", "terminal", "", "rm -rf /private/path", 5))
database.commit()
database.close()
steps = extract_public_execution_steps(str(path), "lineup-turn:exec_001")
self.assertEqual(["检查 qmtbridge health", "获取服务接口说明", "查询行情接口"], [step["title"] for step in steps])
rendered = str(steps)
self.assertNotIn("private-host", rendered)
self.assertNotIn("secret", rendered)
self.assertNotIn("/private", rendered)
self.assertNotIn("rm -rf", rendered)
def test_missing_or_unrecognized_ledger_never_invents_a_step(self) -> None:
self.assertEqual([], extract_public_execution_steps("/missing/state.db", "lineup-turn:exec_001"))
+287
View File
@@ -0,0 +1,287 @@
from __future__ import annotations
import json
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from lineup.protocol import APP_CALL, APP_RESULT, ARTIFACT_OFFER, CLIENT_INVENTORY, EXECUTION, TEXT, TOOL_CALL, TOOL_INVOKE, TOOL_RESULT, UI_EVENT, UI_OPEN, UI_PATCH, UI_CLOSE, agent_output_message, app_call_message, app_result_message, execution_message, extract_user_input, parse_agent_tool_call, parse_agent_tool_invoke, parse_app_result, parse_client_inventory, parse_tool_response, text_message, ui_event_message, ui_open_message, validate_app_call_payload, validate_artifact_offer_payload, validate_tool_response
class ProtocolTest(unittest.TestCase):
@staticmethod
def _inventory() -> dict[str, object]:
value = parse_client_inventory(json.dumps({"v": 1, "type": CLIENT_INVENTORY, "payload": {
"revision": "catalog-1", "standard_components": [], "applications": [], "tools": [],
"capabilities": [{"name": "app.open_url", "version": "1.0", "risk": "user_confirmation", "input_schema": {"type": "object", "required": ["url"], "properties": {"url": "string"}}, "foreground_required": True}],
}}))
assert value is not None
return value
def test_client_inventory_is_bounded_declarative_context(self) -> None:
raw = json.dumps({"v": 1, "type": CLIENT_INVENTORY, "payload": {
"revision": "catalog-1", "standard_components": [{"id": "choice", "version": "1"}],
"applications": [{"id": "lineup.task-dashboard", "version": "0.1.0", "surfaces": [{"id": "task-dashboard", "events": ["cancel"]}]}],
"tools": [{"app_scope": "pomodoro", "tool_id": "pomodoro.start", "description": "开始专注", "input_schema": {"type": "object"}, "handling": "operation", "delivery": "runtime", "activation_requirement": "foreground_required"}],
"capabilities": [{"name": "app.open_url", "version": "1.0", "risk": "user_confirmation", "input_schema": {"type": "object", "required": ["url"], "properties": {"url": "string"}}, "foreground_required": True}],
}})
value = parse_client_inventory(raw)
self.assertEqual("catalog-1", value["revision"] if value else None)
self.assertEqual("pomodoro.start", value["tools"][0]["tool_id"] if value else None)
self.assertIsNone(parse_client_inventory(raw.replace("catalog-1", "x" * 129)))
def test_inventory_rejects_tools_with_unsafe_extra_fields(self) -> None:
raw = json.dumps({"v": 1, "type": CLIENT_INVENTORY, "payload": {
"revision": "catalog-1", "standard_components": [], "applications": [], "tools": [{
"app_scope": "pomodoro", "tool_id": "pomodoro.start", "description": "开始专注", "input_schema": {"type": "object"},
"handling": "operation", "delivery": "runtime", "activation_requirement": "foreground_required", "handler_key": "bad",
}], "capabilities": [],
}})
self.assertIsNone(parse_client_inventory(raw))
def test_execution_trace_is_adapter_shaped_and_has_no_raw_operation_fields(self) -> None:
value = json.loads(execution_message("exec_001", "completed", [{"id": "step_01", "title": "检查 qmtbridge health", "status": "completed", "command": "curl secret"}], conversation_id="conv-1", sender_id="agent"))
self.assertEqual(EXECUTION, value["type"])
self.assertEqual({"id": "step_01", "title": "检查 qmtbridge health", "status": "completed"}, value["payload"]["steps"][0])
def test_round_trip_text_envelope(self) -> None:
payload = text_message("你好", conversation_id="conv-1", sender_id="agent")
value = json.loads(payload)
self.assertEqual(TEXT, value["type"])
self.assertEqual("你好", value["payload"]["text"])
self.assertEqual("conv-1", value["conversation_id"])
def test_final_text_can_carry_only_its_execution_association(self) -> None:
value = json.loads(agent_output_message("天气已查询。", conversation_id="conv-1", sender_id="agent", execution_id="exec_001"))
self.assertEqual("exec_001", value["payload"]["execution_id"])
self.assertEqual("天气已查询。", value["payload"]["text"])
def test_model_shaped_text_is_rebuilt_with_adapter_execution_association(self) -> None:
value = json.loads(agent_output_message(json.dumps({"v": 1, "type": TEXT, "payload": {"text": "已分析", "untrusted": "discard"}}), conversation_id="conv-1", sender_id="agent", execution_id="exec_001"))
self.assertEqual({"text": "已分析", "execution_id": "exec_001"}, value["payload"])
def test_extracts_lineup_text(self) -> None:
raw = json.dumps({"type": TEXT, "conversation_id": "conv-1", "payload": {"text": "hello"}})
self.assertEqual(("hello", "conv-1"), extract_user_input(raw))
def test_extracts_structured_tool_result(self) -> None:
raw = json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {"call_id": "call-1", "status": "completed", "result": {}}})
text, conversation = extract_user_input(raw)
self.assertEqual("LineUp standard interaction response received.", text)
self.assertIsNone(conversation)
def test_ui_surface_messages_are_versioned_and_keep_the_instance_id(self) -> None:
opened = json.loads(
ui_open_message(
"surface_weather_1",
{"id": "weather", "version": "1.0.0", "html": "<main>Weather</main>"},
conversation_id="conv-1",
sender_id="agent",
state={"city": "北京"},
)
)
self.assertEqual(UI_OPEN, opened["type"])
self.assertEqual("surface_weather_1", opened["payload"]["instance_id"])
self.assertEqual("北京", opened["payload"]["state"]["city"])
event = json.loads(
ui_event_message(
"surface_weather_1", "refresh", {"unit": "c"},
conversation_id="conv-1", sender_id="user-1", target_id="agent",
)
)
self.assertEqual(UI_EVENT, event["type"])
self.assertEqual("human", event["sender"]["kind"])
self.assertEqual("refresh", event["payload"]["event"])
def test_app_capability_call_and_result_keep_their_call_id(self) -> None:
call = json.loads(
app_call_message(
"call-1", "app.open_url", {"url": "https://example.com"},
conversation_id="conv-1", sender_id="agent", target_id="user-1", reason="打开文档",
)
)
self.assertEqual(APP_CALL, call["type"])
self.assertEqual("call-1", call["payload"]["call_id"])
result = json.loads(
app_result_message(
"call-1", "completed", conversation_id="conv-1", sender_id="user-1", target_id="agent", result={"opened": True},
)
)
self.assertEqual(APP_RESULT, result["type"])
self.assertTrue(result["payload"]["result"]["opened"])
readable, conversation = extract_user_input(json.dumps(result))
self.assertEqual("conv-1", conversation)
self.assertEqual("LineUp app capability result received.", readable)
def test_agent_app_call_requires_current_inventory_and_strict_shape(self) -> None:
expires_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat()
raw = json.dumps({"v": 1, "type": APP_CALL, "payload": {
"call_id": "app-call-1", "capability": "app.open_url", "reason": "打开文档", "expires_at": expires_at,
"arguments": {"url": "https://example.com"},
}})
accepted = json.loads(agent_output_message(raw, conversation_id="conv-1", sender_id="agent", inventory=self._inventory()))
self.assertEqual(APP_CALL, accepted["type"])
self.assertEqual({"call_id", "capability", "reason", "expires_at", "arguments"}, set(accepted["payload"]))
self.assertEqual(TEXT, json.loads(agent_output_message(raw, conversation_id="conv-1", sender_id="agent"))["type"])
self.assertIsNone(validate_app_call_payload({"call_id": "app-call-1", "capability": "app.open_url", "reason": "打开", "expires_at": expires_at, "arguments": {"url": "https://example.com", "x": "bad"}}, self._inventory()))
self.assertIsNone(validate_app_call_payload({"call_id": "app-call-1", "capability": "app.open_url", "reason": "打开 https://example.com", "expires_at": expires_at, "arguments": {"url": "https://example.com"}}, self._inventory()))
self.assertIsNone(validate_app_call_payload({"call_id": "app-call-1", "capability": "clipboard.write", "reason": "", "expires_at": expires_at, "arguments": {"text": "x"}}, self._inventory()))
def test_agent_tool_invoke_requires_current_inventory_and_strict_shape(self) -> None:
inventory = parse_client_inventory(json.dumps({"v": 1, "type": CLIENT_INVENTORY, "payload": {
"revision": "catalog-1", "standard_components": [], "applications": [], "tools": [
{"app_scope": "pomodoro", "tool_id": "pomodoro.start", "description": "开始专注", "input_schema": {"type": "object"}, "handling": "operation", "delivery": "runtime", "activation_requirement": "foreground_required"}
], "capabilities": [],
}}))
self.assertIsNotNone(inventory)
raw = json.dumps({"v": 1, "type": TOOL_INVOKE, "payload": {
"call_id": "tool-call-1", "inventory_revision": "catalog-1", "app_scope": "pomodoro", "tool_id": "pomodoro.start",
"input": {"duration_seconds": 300},
}})
accepted = json.loads(agent_output_message(raw, conversation_id="conv-1", sender_id="agent", inventory=inventory))
self.assertEqual(TOOL_INVOKE, accepted["type"])
self.assertEqual("pomodoro.start", accepted["payload"]["tool_id"])
self.assertEqual(TEXT, json.loads(agent_output_message(raw, conversation_id="conv-1", sender_id="agent"))["type"])
self.assertIsNotNone(parse_agent_tool_invoke(raw, inventory))
self.assertIsNone(parse_agent_tool_invoke(json.dumps({"v": 1, "type": TOOL_INVOKE, "payload": {
"call_id": "tool-call-1", "inventory_revision": "catalog-1", "app_scope": "pomodoro", "tool_id": "pomodoro.start",
"input": {}, "instance_id": "bad",
}}), inventory))
self.assertIsNone(parse_agent_tool_invoke(json.dumps({"v": 1, "type": TOOL_INVOKE, "payload": {
"call_id": "tool-call-1", "inventory_revision": "wrong", "app_scope": "pomodoro", "tool_id": "pomodoro.start",
"input": {},
}}), inventory))
def test_app_result_is_reduced_before_it_can_reach_the_model(self) -> None:
safe = json.dumps({"v": 1, "type": APP_RESULT, "payload": {"call_id": "app-call-1", "capability": "clipboard.write", "status": "completed", "completed": True}})
self.assertEqual({"call_id": "app-call-1", "capability": "clipboard.write", "status": "completed", "completed": True}, parse_app_result(safe))
unsafe = json.dumps({"v": 1, "type": APP_RESULT, "payload": {"call_id": "app-call-1", "capability": "clipboard.write", "status": "completed", "clipboard": "secret"}})
self.assertIsNone(parse_app_result(unsafe))
def test_agent_artifact_content_is_bounded_and_has_no_url_or_path(self) -> None:
payload = {"artifact_id": "report_001", "name": "report.txt", "mime_type": "text/plain", "size_bytes": 5, "integrity": "verified", "created_at": datetime.now(timezone.utc).isoformat(), "local_ref": "cache-report-001", "content_base64": "aGVsbG8="}
canonical = validate_artifact_offer_payload(payload)
self.assertEqual("aGVsbG8=", canonical["content_base64"] if canonical else None)
promoted = json.loads(agent_output_message(json.dumps({"v": 1, "type": ARTIFACT_OFFER, "payload": payload}), conversation_id="conv-1", sender_id="agent"))
self.assertEqual(ARTIFACT_OFFER, promoted["type"])
self.assertEqual(TEXT, json.loads(agent_output_message(json.dumps({"v": 1, "type": ARTIFACT_OFFER, "payload": {**payload, "url": "https://bad.example"}}), conversation_id="conv-1", sender_id="agent"))["type"])
self.assertIsNone(validate_artifact_offer_payload({**payload, "size_bytes": 4}))
def test_allows_only_valid_standard_tool_calls_from_the_agent(self) -> None:
expires_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat()
choice = json.loads(agent_output_message(
json.dumps({"v": 1, "type": TOOL_CALL, "payload": {
"call_id": "call-choice", "tool": "choice", "title": "环境", "prompt": "选择环境", "expires_at": expires_at,
"data": {"action_group": {"mode": "single-choice", "actions": [{"id": "staging", "label": "Staging"}]}, "html": "<script>bad</script>"},
}}),
conversation_id="conv-1", sender_id="agent", target_id="user-1",
))
self.assertEqual(TOOL_CALL, choice["type"])
self.assertEqual("choice", choice["payload"]["tool"])
self.assertNotIn("html", choice["payload"]["data"])
self.assertEqual("agent", choice["sender"]["kind"])
self.assertEqual("conv-1", choice["conversation_id"])
confirm = parse_agent_tool_call(json.dumps({"v": 1, "type": TOOL_CALL, "payload": {
"call_id": "call-confirm", "tool": "confirm", "title": "确认", "prompt": "继续吗", "expires_at": expires_at,
"data": {"confirm": {"approve_label": "继续", "cancel_label": "停止"}},
}}))
self.assertIsNotNone(confirm)
form = parse_agent_tool_call(json.dumps({"v": 1, "type": TOOL_CALL, "payload": {
"call_id": "call-input", "tool": "input", "title": "版本", "prompt": "输入版本", "expires_at": expires_at,
"data": {"form": {"fields": [{"id": "version", "label": "Version", "type": "text", "required": True, "max_length": 32}], "submit_label": "提交", "cancel_label": "取消"}},
}}))
self.assertIsNotNone(form)
self.assertIsNone(parse_agent_tool_call(json.dumps({"v": 1, "type": TOOL_CALL, "payload": {
"call_id": "call-multi", "tool": "choice", "title": "多选", "prompt": "选两项", "expires_at": expires_at,
"data": {"action_group": {"mode": "multi-choice", "actions": [{"id": "a", "label": "A"}, {"id": "b", "label": "B"}]}},
}})))
def test_invalid_tool_or_surface_output_safely_degrades_to_text(self) -> None:
invalid = json.loads(agent_output_message(
json.dumps({"v": 1, "type": TOOL_CALL, "payload": {"call_id": "call-1", "tool": "shell", "data": {}}}),
conversation_id="conv-1", sender_id="agent",
))
self.assertEqual(TEXT, invalid["type"])
blocked_surface = json.loads(agent_output_message(
json.dumps({"v": 1, "type": UI_OPEN, "payload": {"instance_id": "surface-1", "app": {"app_id": "lineup.task-dashboard", "version": "0.1.0"}, "bundle_url": "https://bad.example/x.js"}}),
conversation_id="conv-1", sender_id="agent",
))
self.assertEqual(TEXT, blocked_surface["type"])
inventory = parse_client_inventory(json.dumps({"v": 1, "type": CLIENT_INVENTORY, "payload": {
"revision": "catalog-1", "standard_components": [], "applications": [], "tools": [], "capabilities": [],
}}))
blocked_invoke = json.loads(agent_output_message(
json.dumps({"v": 1, "type": TOOL_INVOKE, "payload": {"call_id": "call-1", "inventory_revision": "catalog-1", "app_scope": "pomodoro", "tool_id": "pomodoro.start", "input": {}}}),
conversation_id="conv-1", sender_id="agent", inventory=inventory,
))
self.assertEqual(TEXT, blocked_invoke["type"])
def test_allows_only_the_canonical_m2_task_surface_lifecycle(self) -> None:
opened = json.loads(agent_output_message(json.dumps({"v": 1, "type": UI_OPEN, "payload": {"instance_id": "surface-1", "app": {"app_id": "lineup.task-dashboard", "version": "0.1.0"}, "state": {"title": "Build", "percent": 25}}}), conversation_id="conv-1", sender_id="agent"))
patched = json.loads(agent_output_message(json.dumps({"v": 1, "type": UI_PATCH, "payload": {"instance_id": "surface-1", "state": {"percent": 60}}}), conversation_id="conv-1", sender_id="agent"))
closed = json.loads(agent_output_message(json.dumps({"v": 1, "type": UI_CLOSE, "payload": {"instance_id": "surface-1"}}), conversation_id="conv-1", sender_id="agent"))
self.assertEqual(UI_OPEN, opened["type"])
self.assertEqual({"app_id": "lineup.task-dashboard", "version": "0.1.0"}, opened["payload"]["app"])
self.assertEqual(UI_PATCH, patched["type"])
self.assertEqual(UI_CLOSE, closed["type"])
def test_tool_response_is_schema_bound_and_does_not_preserve_ui_injection(self) -> None:
raw = json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-choice", "status": "completed", "result": {"action_ids": ["staging"], "ui": {"html": "<script>bad</script>"}},
}})
response = parse_tool_response(raw)
self.assertIsNotNone(response)
self.assertEqual(
{"call_id": "call-choice", "outcome": "completed", "action_ids": ["staging"]},
validate_tool_response(response, {"tool": "choice", "mode": "single-choice", "action_ids": ["staging"]}),
)
clean = parse_tool_response(json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-choice", "status": "completed", "result": {"action_ids": ["staging"]},
}}))
self.assertEqual(
{"call_id": "call-choice", "outcome": "completed", "action_ids": ["staging"]},
validate_tool_response(clean, {"tool": "choice", "mode": "single-choice", "action_ids": ["staging"]}),
)
def test_single_choice_accepts_host_action_id_shape(self) -> None:
raw = json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-choice", "status": "completed", "result": {"action_id": "staging"},
}})
response = parse_tool_response(raw)
self.assertIsNotNone(response)
self.assertEqual(
{"call_id": "call-choice", "outcome": "completed", "action_ids": ["staging"]},
validate_tool_response(response, {"tool": "choice", "mode": "single-choice", "action_ids": ["staging"]}),
)
self.assertIsNone(validate_tool_response(
parse_tool_response(json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-choice", "status": "completed", "result": {"action_id": "unknown"},
}})),
{"tool": "choice", "mode": "single-choice", "action_ids": ["staging"]},
))
def test_single_field_input_accepts_host_text_shape(self) -> None:
raw = json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-input", "status": "completed", "result": {"text": "验收 04A other-input"},
}})
response = parse_tool_response(raw)
self.assertIsNotNone(response)
self.assertEqual(
{"call_id": "call-input", "outcome": "completed", "values": {"task": "验收 04A other-input"}},
validate_tool_response(response, {"tool": "input", "fields": {"task": {"type": "text", "required": True, "min_length": None, "max_length": None}}}),
)
self.assertIsNone(validate_tool_response(
parse_tool_response(json.dumps({"v": 1, "type": TOOL_RESULT, "payload": {
"call_id": "call-input", "status": "completed", "result": {"text": 1},
}})),
{"tool": "input", "fields": {"task": {"type": "text", "required": True, "min_length": None, "max_length": None}}},
))
if __name__ == "__main__":
unittest.main()
+69
View File
@@ -0,0 +1,69 @@
from __future__ import annotations
import sys
import tempfile
import time
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from lineup.state import StateStore
class StateStoreTest(unittest.TestCase):
def test_claim_is_idempotent_and_running_work_is_recovered(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
message = {"message_id": 7, "message_seq": 4, "from_uid": "user-1", "payload": "hello"}
self.assertTrue(store.claim_inbox(message))
self.assertFalse(store.claim_inbox(message))
row = store.pending()[0]
store.set_inbox_state(row["message_id"], "running")
store.recover_interrupted_turns()
self.assertEqual("pending", store.pending()[0]["state"])
store.close()
def test_outbox_persists_until_acknowledged(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
item_id = store.enqueue_outbox("payload")
self.assertEqual(1, len(store.outbox_items()))
store.mark_outbox_sent(item_id)
self.assertEqual([], store.outbox_items())
store.close()
def test_bootstrap_marker_is_persistent(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
self.assertFalse(store.is_bootstrapped())
store.set_bootstrapped()
self.assertTrue(store.is_bootstrapped())
store.close()
def test_tool_responses_are_idempotent_and_scoped_to_a_conversation(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
schema = {"tool": "choice", "mode": "single-choice", "action_ids": ["staging"]}
self.assertEqual("accepted", store.register_tool_call("conv-a", "call-1", "choice", time.time() + 60, schema))
self.assertEqual("accepted", store.register_tool_call("conv-b", "call-1", "choice", time.time() + 60, schema))
self.assertEqual("accepted", store.consume_tool_response("conv-a", "call-1", "result"))
self.assertEqual("duplicate", store.consume_tool_response("conv-a", "call-1", "result"))
self.assertEqual("accepted", store.consume_tool_response("conv-b", "call-1", "cancel"))
audit = [(row["event"], row["disposition"]) for row in store.tool_audit("conv-a", "call-1")]
self.assertEqual([("call", "accepted"), ("result", "accepted"), ("result", "duplicate")], audit)
store.close()
def test_expired_tool_response_is_audited_without_being_accepted(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
store = StateStore(Path(temp_dir) / "state.sqlite3")
store.register_tool_call("conv-a", "call-expired", "confirm", time.time() - 1, {"tool": "confirm"})
self.assertEqual("expired", store.consume_tool_response("conv-a", "call-expired", "result"))
row = store.tool_call("conv-a", "call-expired")
self.assertEqual("expired", row["state"])
self.assertEqual(("result", "expired"), tuple(store.tool_audit("conv-a", "call-expired")[-1]))
store.close()
if __name__ == "__main__":
unittest.main()