Agent runtime

# src/lerim/runtime/agent.py:166-201
class LerimAgent:
    """Lead runtime wrapper for chat, sync, and maintain orchestration."""

    def __init__(
        self,
        model: str | None = None,
        provider: str | None = None,
        timeout_seconds: int | None = None,
        single_tools: list[str] | None = None,
        allowed_read_dirs: list[str | Path] | None = None,
        default_cwd: str | None = None,
    ) -> None:
        """Create runtime with role-based model wiring and tool policy defaults."""
        config = get_config()
        lead_role = config.lead_role
        # ... role configuration ...
        self.model = build_orchestration_model_from_role(role, config=config)
        self.system_prompt = build_lead_system_prompt()

# src/lerim/runtime/agent.py:216-242
def _build_lead_agent(self, mode: str) -> Agent[RuntimeToolContext, str]:
    """Build one lead PydanticAI agent with mode-specific tool registration."""
    mode_tools = {
        "ask": CHAT_TOOLS,        # ["read", "grep", "glob", "explore"]
        "sync": SYNC_TOOLS,       # ask tools + "write", "write_memory", "extract_pipeline", "summarize_pipeline"
        "maintain": MAINTAIN_TOOLS,  # ask tools + "write", "write_memory", "edit"
    }
    allowed_tools = set(
        self.single_tools if mode == "ask" else mode_tools.get(mode, CHAT_TOOLS)
    )

    agent = Agent[
        RuntimeToolContext,
        str,
    ](
        model=self.model,
        output_type=str,
        deps_type=RuntimeToolContext,
        name=f"lerim-{mode}",
        instructions=instructions,
        retries=retries,
        tool_timeout=float(self._timeout_seconds),
    )

# src/lerim/runtime/agent.py:244-259
@agent.tool
def read(
    ctx: RunContext[RuntimeToolContext],
    file_path: str,
    offset: int = 1,
    limit: int = 2000,
) -> str:
    """Read a file and return numbered lines. 
    If file_path is a directory, list its entries instead. 
    Use offset/limit to paginate large files."""
    return read_file_tool(
        context=ctx.deps,
        file_path=file_path,
        offset=offset,
        limit=limit,
    )

# src/lerim/runtime/agent.py:326-346
@agent.tool
def write_memory(
    ctx: RunContext[RuntimeToolContext],
    primitive: str,
    title: str,
    body: str,
    confidence: float = 0.8,
    tags: list[str] | None = None,
    kind: str | None = None,
) -> dict[str, Any]:
    """Write a structured memory record. 
    Python builds the markdown — pass fields directly. 
    primitive must be 'decision' or 'learning'. 
    kind is required for learnings (insight/procedure/friction/pitfall/preference)."""
    return write_memory_tool(
        context=ctx.deps,
        primitive=primitive,
        title=title,
        body=body,
        confidence=confidence,
        tags=tags,
        kind=kind,
    )

# src/lerim/runtime/agent.py:367-391
@agent.tool
def extract_pipeline(
    ctx: RunContext[RuntimeToolContext],
    guidance: str | None = None,
) -> dict[str, Any]:
    """Run DSPy extraction pipeline on the session trace. 
    Paths and metadata are handled automatically. 
    Pass optional guidance about focus areas or dedupe hints."""
    return run_extract_pipeline_tool(
        context=ctx.deps,
        guidance=guidance,
    )

@agent.tool
def summarize_pipeline(
    ctx: RunContext[RuntimeToolContext],
    guidance: str | None = None,
) -> dict[str, Any]:
    """Run DSPy summarization pipeline on the session trace. 
    Paths and metadata are handled automatically. 
    Pass optional guidance about focus areas."""
    return run_summarization_pipeline_tool(
        context=ctx.deps,
        guidance=guidance,
    )

# src/lerim/runtime/agent.py:293-308
@agent.tool
async def explore(
    ctx: RunContext[RuntimeToolContext],
    query: str,
) -> dict[str, Any]:
    """Delegate read-only evidence gathering to explorer subagent.

    Async so PydanticAI can run multiple explore calls in parallel
    when the LLM emits them in the same tool-call turn (max 4).
    """
    result = await get_explorer_agent().run(
        query, deps=ctx.deps, usage=ctx.usage
    )
    return result.output.model_dump()

# src/lerim/runtime/subagents.py:21-44
def _build_explorer(model=None) -> Agent[RuntimeToolContext, ExplorerEnvelope]:
    """Build read-only explorer subagent with glob/read/grep tools."""
    agent = Agent[RuntimeToolContext, ExplorerEnvelope](
        model=model or build_orchestration_model("explorer"),
        output_type=ExplorerEnvelope,
        deps_type=RuntimeToolContext,
        name="lerim-explorer",
        instructions="""
You are a read-only explorer for Lerim memories and workspace artifacts.

Memory layout (base_path defaults to the memory root passed by the lead agent):
- decisions/*.md — architecture/design decisions
- learnings/*.md — insights, procedures, facts
- summaries/YYYYMMDD/HHMMSS/*.md — session summaries

Each .md file has YAML frontmatter (id, title, confidence, tags, kind, created) 
then a markdown body.

Search strategy:
1. Use grep to find memories by keyword, title, or tag.
2. Use glob to list files when you need to scan a directory.
3. Use read to get the full content of specific files found by grep/glob.
Return structured evidence with file paths.""",
        retries=1,
    )

# src/lerim/runtime/tools.py:132-149
def _write_allowed_roots(context: RuntimeToolContext) -> tuple[Path, ...]:
    """Return allowed write roots for write/edit tools."""
    roots: list[Path] = []
    if context.memory_root:
        roots.append(context.memory_root)
    if context.run_folder:
        roots.append(context.run_folder)
    return tuple(dict.fromkeys(roots))

def _assert_write_boundary(path: Path, context: RuntimeToolContext) -> None:
    """Raise when write target is outside memory/run-folder boundaries."""
    roots = _write_allowed_roots(context)
    if not roots or not any(_is_within(path, root) for root in roots):
        raise RuntimeError(
            f"Cannot write '{path}': outside allowed roots. "
            f"Writable paths: {', '.join(str(r) for r in roots)}"
        )

# src/lerim/runtime/tools.py:294-324
def write_file_tool(
    *,
    context: RuntimeToolContext,
    file_path: str,
    content: str,
) -> dict[str, Any]:
    """Write file content under guarded roots. 
    Memory primitives are rejected — use write_memory_tool instead."""
    resolved = _resolve_path(file_path, _default_cwd(context))
    _assert_write_boundary(resolved, context)
    
    primitive_type = _memory_primitive_type(resolved, context.memory_root)
    if primitive_type is not None and primitive_type != MemoryType.summary:
        raise ModelRetry(
            "Use write_memory tool for memory files. "
            "It accepts structured fields (primitive, title, body, confidence, tags, kind) "
            "and builds the markdown automatically."
        )
    if primitive_type == MemoryType.summary:
        raise ModelRetry(
            "Cannot write summaries directly. Use summarize_pipeline tool."
        )

# src/lerim/runtime/tools.py:107-129
def _read_allowed_roots(context: RuntimeToolContext) -> tuple[Path, ...]:
    """Return allowed read roots for read/glob/grep tools."""
    roots: list[Path] = []
    if context.memory_root:
        roots.append(context.memory_root)
    if context.workspace_root:
        roots.append(context.workspace_root)
    if context.run_folder:
        roots.append(context.run_folder)
    roots.append(_global_cache_dir())  # ~/.lerim/cache
    roots.extend(context.extra_read_roots)
    return tuple(dict.fromkeys(roots))

def _check_read_boundary(path: Path, context: RuntimeToolContext) -> str | None:
    """Return error string when read target is outside approved roots, else None."""
    roots = _read_allowed_roots(context)
    if not roots or not any(_is_within(path, root) for root in roots):
        return (
            f"ERROR: Cannot read '{path}': outside allowed roots. "
            f"Readable paths: {', '.join(str(r) for r in roots)}"
        )
    return None

# src/lerim/runtime/agent.py:491-608
def sync(
    self,
    trace_path: str | Path,
    memory_root: str | Path | None = None,
    workspace_root: str | Path | None = None,
) -> dict[str, Any]:
    """Run memory-write sync flow and return stable contract payload."""
    trace_file = Path(trace_path).expanduser().resolve()
    if not trace_file.exists() or not trace_file.is_file():
        raise FileNotFoundError(f"trace_path_missing:{trace_file}")

    # ... resolve roots and create run folder ...
    
    prompt = build_sync_prompt(
        trace_file=trace_file,
        memory_root=resolved_memory_root,
        run_folder=run_folder,
        artifact_paths=artifact_paths,
        metadata=metadata,
    )
    
    context = build_tool_context(
        repo_root=repo_root,
        memory_root=resolved_memory_root,
        workspace_root=resolved_workspace_root,
        run_folder=run_folder,
        extra_read_roots=extra_roots,
        run_id=run_folder.name,
        config=self.config,
        trace_path=trace_file,
        artifact_paths=artifact_paths,
    )
    
    response, _, cost_usd = self._run_agent_once(
        prompt=prompt,
        mode="sync",
        context=context,
    )
    
    # ... validate artifacts and return contract payload ...

[roles.lead]
provider = "openrouter"
model = "x-ai/grok-4.1-fast"
timeout_seconds = 600
max_iterations = 10
fallback_models = ["anthropic/claude-3.7-sonnet"]

[roles.explorer]
provider = "openrouter"
model = "x-ai/grok-4.1-fast"