mirror of
https://github.com/andrewyng/openworker.git
synced 2026-09-04 07:20:10 +00:00
Rohit C Prasad
f6ad97274d
Stream drops between chunks (+ the pre-first-token wait); pending approvals/questions/plans resolve as interrupted. Running shell commands die via an executor interrupt hook; skipped tool calls still get results (no orphans). Also fixes delete_session calling a nonexistent engine.interrupt().
340 lines
15 KiB
Python
340 lines
15 KiB
Python
"""Engine assembly from an Agent (Code / Chat / …).
|
|
|
|
Wires the agent's base tools + permissions + AGENTS.md (workspace agents) + memory +
|
|
the skill catalog (progressive disclosure) + load_skill into a TurnEngine.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from .agents import Agent, AgentContext, code_agent
|
|
from .automation import scheduling_tools
|
|
from .selfwake import selfwake_tools
|
|
from .subscriptions import subscription_tools
|
|
from .config import load_config
|
|
from .connectors import (
|
|
connector_list,
|
|
load_settings,
|
|
make_integration_tools,
|
|
make_send_file_tool,
|
|
make_send_message_tool,
|
|
)
|
|
from .engine import Approver, TurnEngine
|
|
from .environment import environment_context
|
|
from .memory import MemoryStore, Scope, format_memories, memory_tools
|
|
from .permissions import Mode, PermissionEngine
|
|
from .project import load_agents_md
|
|
from .roots import RootDir, normalize_roots, render_context
|
|
from .providers import ProviderClient, ProviderRouter
|
|
from .overrides import RiskOverrideStore
|
|
from .secrets import SecretStore, state_dir
|
|
from .skills import SkillLoader, skill_catalog_text, skill_tools
|
|
from .tools import ToolRegistry
|
|
from .tools.ask import ask_user_tool
|
|
from .tools.directories import request_directory_tool
|
|
from .tools.plan import propose_plan_tool
|
|
from .tools.subagent import explorer_tools
|
|
from .web import make_web_fetch_tool, make_web_search_tool
|
|
from .tools.shell import LocalExecutor
|
|
from .tools.todo import TodoList
|
|
|
|
# Appended each turn while discuss mode is active: enforcement-only read-only, with no
|
|
# pressure toward a plan proposal (that's what distinguishes it from plan mode).
|
|
_DISCUSS_MODE_CONTEXT = """\
|
|
Discuss mode is active: write and shell tools are disabled. Explore and answer freely; if
|
|
the user asks for a change, describe it in chat instead of attempting it (they can switch
|
|
to plan or approval mode to have you make it)."""
|
|
|
|
# Appended to the latest user message every turn while plan mode is active. The mode can
|
|
# flip mid-session (plan approval), so this can't live in the static instructions.
|
|
_PLAN_MODE_CONTEXT = """\
|
|
Plan mode is active: write and shell tools are blocked. Explore read-only and design an
|
|
approach. When you've committed to one, present it with `propose_plan` (what you'll change,
|
|
in which files, how you'll verify) — don't describe edits as if you were making them. If
|
|
the plan is approved, this same session switches to execution and you implement it; if
|
|
rejected, revise the plan using the feedback."""
|
|
|
|
# When-to-remember rules, injected only when a memory store is wired. Without these,
|
|
# models either never call `remember` or save noise the repo already records.
|
|
_MEMORY_GUIDANCE = """\
|
|
Memory:
|
|
- You have persistent memory across sessions. Use `remember` for durable facts: the user's \
|
|
corrections and stated preferences (include the why), and project context you couldn't \
|
|
rederive from the code. Don't save what the repo already records (code structure, git \
|
|
history, AGENTS.md) or details that only matter to the current task. Use absolute dates, \
|
|
never "yesterday".
|
|
- Before saving, check the known-memories list: if an entry already covers it, revise that \
|
|
entry with `memory_update` instead of adding a near-duplicate; retire wrong or obsolete \
|
|
entries with `memory_forget`.
|
|
- Memories reflect when they were written. If one names a file, flag, or URL, verify it \
|
|
still exists before relying on it."""
|
|
|
|
# UX-015 (§33): the GUI interleaves these status lines with humanized tool rows inside a
|
|
# collapsed "turn" — they're what the user reads while the agent works. Universal (appended
|
|
# for every persona); models that ignore it degrade gracefully to a turn with no narration.
|
|
_NARRATION_GUIDANCE = """\
|
|
Narration: before each batch of tool calls, write ONE short plain sentence saying what \
|
|
you're doing and why (e.g. "Checking what merged since yesterday's digest."). It is shown \
|
|
to the user as live progress. Don't narrate trivial single-call follow-ups, don't repeat \
|
|
the previous line, and never let narration replace your final answer."""
|
|
|
|
|
|
def _enabled_connector_tools(secrets: SecretStore) -> tuple[set[str], set[str]]:
|
|
connectors = {c["name"]: c for c in connector_list(secrets)}
|
|
enabled_connectors = {
|
|
name
|
|
for name, c in connectors.items()
|
|
if c.get("connected") and c.get("enabled")
|
|
}
|
|
enabled_tools = {
|
|
tool["name"]
|
|
for c in connectors.values()
|
|
if c.get("name") in enabled_connectors
|
|
for tool in c.get("tools", [])
|
|
if tool.get("enabled")
|
|
}
|
|
return enabled_connectors, enabled_tools
|
|
|
|
|
|
def _skill_dirs(workspace: Optional[Path]) -> list[Path]:
|
|
dirs = [state_dir() / "skills"]
|
|
if workspace is not None:
|
|
dirs.append(workspace / ".coworker" / "skills")
|
|
return dirs
|
|
|
|
|
|
def build_engine(
|
|
*,
|
|
agent: Agent,
|
|
workspace: Optional[str | Path] = None,
|
|
model: str = "gpt-5.6-sol",
|
|
mode: Mode = Mode.INTERACTIVE,
|
|
approver: Optional[Approver] = None,
|
|
provider: Optional[ProviderClient] = None,
|
|
allowed_commands: Optional[list[str]] = None,
|
|
max_iterations: Optional[int] = None,
|
|
model_settings: Optional[dict[str, Any]] = None,
|
|
memory_store: Optional[MemoryStore] = None,
|
|
messages: Optional[list[dict[str, Any]]] = None,
|
|
extra_tools: Optional[list[Any]] = None,
|
|
secrets: Optional[SecretStore] = None,
|
|
task_store: Optional[Any] = None,
|
|
wake_store: Optional[Any] = None,
|
|
session_id: Optional[str] = None,
|
|
audit_sink: Optional[Any] = None,
|
|
roots: Optional[list] = None,
|
|
directory_requester: Optional[Any] = None,
|
|
plan_approver: Optional[Any] = None,
|
|
question_asker: Optional[Any] = None,
|
|
subscription_store: Optional[Any] = None,
|
|
channel_buffer: Optional[Any] = None,
|
|
routing_targets: Optional[list[str]] = None,
|
|
connector_filter: Optional[set[str]] = None,
|
|
) -> TurnEngine:
|
|
ws = Path(workspace).expanduser().resolve() if workspace else None
|
|
if agent.needs_workspace and ws is None:
|
|
raise ValueError(f"agent '{agent.name}' requires a workspace")
|
|
|
|
# The session's directories. Explicit `roots` (orphan Cowork: scratch + added folders) wins;
|
|
# otherwise the single workspace is the sole writable root. One shared, mutable list flows to
|
|
# the file tools, the permission engine, and the context injector so add/remove is seen by all.
|
|
if roots:
|
|
root_list: list[RootDir] = normalize_roots(roots)
|
|
elif ws is not None:
|
|
root_list = [RootDir(path=ws, writable=True)]
|
|
else:
|
|
root_list = []
|
|
|
|
config = load_config(ws)
|
|
executor = (
|
|
LocalExecutor(cwd=ws) if (agent.needs_workspace and ws is not None) else None
|
|
)
|
|
todo = TodoList()
|
|
context = AgentContext(
|
|
workspace=ws, executor=executor, todo=todo, roots=root_list or None
|
|
)
|
|
|
|
registry = ToolRegistry()
|
|
registry.register_all(agent.build_tools(context))
|
|
# MCP / connector tools (supplied by the manager) carry their own metadata + schema.
|
|
if extra_tools:
|
|
registry.register_all(extra_tools)
|
|
# Messaging personas (Cowork / Ops / MyHelper) expose send_message; MyHelper also uses it as
|
|
# the reply path for inbound Telegram/Slack super-agent sessions.
|
|
secrets = secrets or SecretStore()
|
|
if agent.messaging and any(s.enabled for s in load_settings(secrets).values()):
|
|
registry.register(make_send_message_tool(secrets))
|
|
# send_file (§34): hand deliverables into the chat — same targets, but its OWN
|
|
# approval surface (a thread's standing send_message grant never covers uploads).
|
|
registry.register(
|
|
make_send_file_tool(secrets, workspace=ws, roots=root_list or None)
|
|
)
|
|
# Channel subscriptions (inbound): listen to a channel, catch up, (un)subscribe. The agent
|
|
# obtains a channel via ask_user or from a channel message it's reacting to.
|
|
if subscription_store is not None and channel_buffer is not None and session_id:
|
|
registry.register_all(
|
|
subscription_tools(
|
|
subscription_store,
|
|
session_id,
|
|
channel_buffer,
|
|
routing_targets=routing_targets,
|
|
)
|
|
)
|
|
# Knowledge surfaces with a multi-root workspace can ask the user mid-task for another folder.
|
|
if agent.family == "knowledge" and root_list:
|
|
registry.register(request_directory_tool())
|
|
if agent.connectors:
|
|
enabled_connectors, enabled_tools = _enabled_connector_tools(secrets)
|
|
# Per-session connection hierarchy (UI-REFRESH §4.3): when the caller supplies the session's
|
|
# effective connector set, intersect it so only effective-enabled connectors expose tools.
|
|
# Default None preserves CLI / direct callers (no per-session restriction).
|
|
if connector_filter is not None:
|
|
enabled_connectors = enabled_connectors & connector_filter
|
|
registry.register_all(
|
|
make_integration_tools(
|
|
secrets,
|
|
enabled_connectors=enabled_connectors,
|
|
enabled_tools=enabled_tools,
|
|
roots=root_list or None,
|
|
)
|
|
)
|
|
# Web search + fetch: research tools for every agent (keyless DuckDuckGo default).
|
|
registry.register(make_web_search_tool(secrets))
|
|
registry.register(make_web_fetch_tool())
|
|
# ask_user: the universal human-in-the-loop Q&A primitive (every agent; engine-intercepted).
|
|
if question_asker is not None:
|
|
registry.register(ask_user_tool())
|
|
# Route by the model's `provider:` prefix (OpenAI default, Ollama, …). The manager normally
|
|
# passes its shared router; this fallback covers the TUI / direct build_engine() callers.
|
|
# Resolved here (not at engine construction) because the explorer subagent captures it.
|
|
provider = provider or ProviderRouter(secrets, default_provider="openai")
|
|
# Code-family personas can fan broad research out to read-only explorer subagents, keeping
|
|
# their own context for the actual change.
|
|
if agent.family == "code" and ws is not None:
|
|
registry.register_all(
|
|
explorer_tools(
|
|
workspace=ws,
|
|
provider=provider,
|
|
model=model,
|
|
model_settings=model_settings,
|
|
)
|
|
)
|
|
# Scheduling: knowledge surfaces with a workspace can set up scheduled tasks (origin = this
|
|
# session). Code stays out (it fans out to explorers instead).
|
|
if task_store is not None and ws is not None and agent.family == "knowledge":
|
|
origin = {
|
|
"surface": agent.name,
|
|
"session_id": session_id or "",
|
|
"workspace": str(ws),
|
|
"agent": agent.name,
|
|
}
|
|
registry.register_all(
|
|
scheduling_tools(task_store, origin=origin, default_workspace=str(ws))
|
|
)
|
|
# Self-wake: knowledge surfaces can suspend + schedule their own resumption (timer /
|
|
# on-completion / on-event). The scheduler tick resumes due wakes.
|
|
if wake_store is not None and session_id and agent.family == "knowledge":
|
|
registry.register_all(selfwake_tools(wake_store, session_id))
|
|
|
|
instructions = f"{agent.system_prompt}\n\n{_NARRATION_GUIDANCE}"
|
|
if ws is not None:
|
|
instructions = f"{instructions}\n\n{environment_context(ws)}"
|
|
conventions = load_agents_md(ws)
|
|
if conventions:
|
|
instructions = f"{instructions}\n\n{conventions}"
|
|
|
|
if memory_store is not None:
|
|
registry.register_all(
|
|
memory_tools(memory_store, workspace=str(ws) if ws else None)
|
|
)
|
|
instructions = f"{instructions}\n\n{_MEMORY_GUIDANCE}"
|
|
remembered = memory_store.list(scope=Scope.GLOBAL)
|
|
if ws is not None:
|
|
remembered += memory_store.list(scope=Scope.WORKSPACE, workspace=str(ws))
|
|
block = format_memories(remembered)
|
|
if block:
|
|
instructions = f"{instructions}\n\n{block}"
|
|
|
|
skill_loader = SkillLoader(_skill_dirs(ws))
|
|
registry.register_all(skill_tools(skill_loader))
|
|
catalog = skill_catalog_text(skill_loader)
|
|
if catalog:
|
|
instructions = f"{instructions}\n\n{catalog}"
|
|
|
|
# User-local risk overrides (mainly to relax MCP's conservative default). Empty store →
|
|
# no-op; never written by persona loading (the no-self-grant rule).
|
|
risk_overrides = RiskOverrideStore(state_dir() / "risk_overrides.json").resolver()
|
|
permissions = PermissionEngine(
|
|
workspace_root=ws or (root_list[0].path if root_list else Path.cwd()),
|
|
mode=mode,
|
|
allowed_commands=allowed_commands or config.allowed_commands,
|
|
auto_allow_tools=set(config.auto_allow),
|
|
roots=root_list or None,
|
|
risk_overrides=risk_overrides,
|
|
)
|
|
# The plan-mode exit door. Always registered (surfaces can flip a live session into
|
|
# plan mode via set_mode, and the registry is fixed at build); the engine rejects the
|
|
# call whenever the session isn't actually in plan mode.
|
|
registry.register(propose_plan_tool())
|
|
|
|
# Per-turn ephemeral context, appended to the latest user message since mid-thread system
|
|
# messages aren't reliable across providers. Two producers: the plan-mode reminder (mode can
|
|
# flip mid-session, so it's checked each turn, not baked into the instructions) and the live
|
|
# directory list (orphan Cowork can gain folders mid-session; Cowork/MyHelper only).
|
|
roots_context = (
|
|
(lambda: render_context(root_list))
|
|
if root_list and agent.family == "knowledge"
|
|
else None
|
|
)
|
|
|
|
def context_provider() -> str:
|
|
parts = []
|
|
if permissions.mode is Mode.PLAN:
|
|
parts.append(_PLAN_MODE_CONTEXT)
|
|
elif permissions.mode is Mode.DISCUSS:
|
|
parts.append(_DISCUSS_MODE_CONTEXT)
|
|
if roots_context is not None:
|
|
ctx = roots_context()
|
|
if ctx:
|
|
parts.append(ctx)
|
|
return "\n\n".join(parts)
|
|
|
|
engine = TurnEngine(
|
|
provider=provider,
|
|
registry=registry,
|
|
permissions=permissions,
|
|
model=model,
|
|
instructions=instructions,
|
|
approver=approver,
|
|
# Stop kills the in-flight foreground shell command, not just the loop.
|
|
interrupt_hooks=[executor.interrupt_now] if executor is not None else None,
|
|
max_iterations=(
|
|
max_iterations if max_iterations is not None else config.max_iterations
|
|
),
|
|
model_settings=model_settings,
|
|
messages=messages,
|
|
audit_sink=audit_sink,
|
|
context_provider=context_provider,
|
|
directory_requester=directory_requester,
|
|
plan_approver=plan_approver,
|
|
question_asker=question_asker,
|
|
)
|
|
engine.executor = executor # type: ignore[attr-defined]
|
|
engine.todo = todo # type: ignore[attr-defined]
|
|
engine.agent_name = agent.name # type: ignore[attr-defined]
|
|
engine.roots = root_list # type: ignore[attr-defined] # shared list; Slice C mutates in place
|
|
engine.audit_context = {
|
|
"session_id": session_id or "",
|
|
"agent": agent.name,
|
|
"workspace": str(ws) if ws else "",
|
|
}
|
|
engine.skill_loader = skill_loader # type: ignore[attr-defined]
|
|
return engine
|
|
|
|
|
|
def build_code_engine(**kwargs: Any) -> TurnEngine:
|
|
"""Back-compat shim: build the Code agent's engine."""
|
|
return build_engine(agent=code_agent(), **kwargs)
|