Commit 4e874f4

**Chose Research Desk before shopping/trading.** Same harness DNA (tools, memory, citations, light HITL) with a vertical that needs no Gmail/Calendar/checkout. Product surface: `operator.yaml` + `ai-agent brief`. Irreversible actions go through `request_approval` rather than a new `AgentDecision.kind`, keeping the decision schema stable.

**Chose a guarded scaffold, not unsupervised self-modification.** Failures are mined into `HarnessPatch` surfaces limited to YAML (`system_prompt` append, `max_tool_rounds`). Accept runs pytest and requires a human CLI step. Arbitrary Python edits and auto-merge are out of v0 — matches “study while touching code” without claiming AGI. Framing and reading list live in the README moonshot section.

- Not an unsupervised self-modifying agent (Self-Harness patches are human-gated config only)

| **Research Desk** | Done | Personal Operator persona + `brief` → cited markdown under `briefs/` | [`config/agents/operator.yaml`](config/agents/operator.yaml) · [`application/brief.py`](src/ai_agent/application/brief.py) |

| **HITL approvals** | Done | `request_approval` tool + optional `brief --approve` (console Y/n) | [`tools/request_approval.py`](src/ai_agent/tools/request_approval.py) · [`infrastructure/approval.py`](src/ai_agent/infrastructure/approval.py) |

| **Self-Harness (experimental)** | Scaffold | Mine failures → propose YAML patches → human `accept` after pytest | [`application/self_harness.py`](src/ai_agent/application/self_harness.py) |

| Research operator | `uv run ai-agent -c config/agents/operator.yaml` | Interactive Research Desk |

| One-shot brief | `uv run ai-agent brief "agent harness"` | Writes `briefs/YYYYMMDD_slug.md` |

| Brief + approve | `uv run ai-agent brief "topic" --approve` | Console Y/n before write |

| Harness propose | `uv run ai-agent harness propose` | Mine `.ai_agent/failures` → `proposals/` |

| Harness accept | `uv run ai-agent harness accept <id>` | Pytest gate, then merge into YAML |

First product vertical on this harness: a **research operator** that turns a topic into a cited brief.

uv run ai-agent brief "agent harness"

uv run ai-agent -c config/agents/operator.yaml # interactive

The operator prefers `web_search` → `http_get` → local `retrieve` / `workspace_search`, remembers prefs via `memory` (`pref.*` keys), and never invents sources. Optional `--approve` gates publication. Irreversible future actions use the `request_approval` tool.

Industry frontier (not “solved”): a fixed model improves the **software around itself** — prompts, tool descriptions, loop budgets — from execution evidence, without weight updates. Canonical loop: weakness mining → harness proposal → validation (held-in improves, held-out does not regress).

**Reading path**

| Resource | Why |

|---|---|

| [Self-Harness (arXiv)](https://arxiv.org/abs/2606.09498) | Core paradigm + results |

| [Lil’Log — Harness Engineering for Self-Improvement](https://lilianweng.github.io/posts/2026-07-04-harness/) | Map of self-improvement vs weight updates |

| [LangChain — Anatomy of an Agent Harness](https://www.langchain.com/blog/the-anatomy-of-an-agent-harness) | Agent = Model + Harness vocabulary |

| [Anthropic — Effective harnesses for long-running agents](https://www.anthropic.com/engineering/effective-harnesses-for-long-running-agents) | Long-horizon reliability |

**Scaffold in this repo (human-gated):** failures auto-log on step exceptions; `harness propose` writes a JSON patch (prompt append / `max_tool_rounds` only); `harness accept` runs a pytest subset then merges into YAML. **No auto-merge. No arbitrary Python edits in v0.**

uv run ai-agent harness record-failure "Timed out waiting for researcher"

uv run ai-agent harness propose

uv run ai-agent harness accept patch_… -c config/agents/operator.yaml

| `request_approval` | Pause for human Y/n before irreversible actions |

model: openai/gpt-4o-mini

system_prompt: >

max_tool_rounds: 8

personality:

tone: professional

style: concise

greeting: "Research desk online. Give me a topic and I'll produce a cited brief."

workspace_root: "."

sqlite_path: ".ai_agent/operator.db"

chroma_path: ".ai_agent/chroma"

tools:

- web_search

- http_get

- retrieve

- workspace_search

- memory

- calculator

- current_time

- request_approval

from ai_agent.application.brief import run_research_brief

from ai_agent.application.self_harness import (

accept_harness_patch,

propose_harness_patch,

record_failure,

)

"accept_harness_patch",

"propose_harness_patch",

"record_failure",

"run_research_brief",

from aioconsole import ainput, aprint

from ai_agent.console_io import restore_blocking_stdio

try:

from ai_agent.application.self_harness import record_failure

record_failure(

agent_id=self.agent_id,

message=str(exc),

tool_traces=[t.model_dump() for t in session.tool_traces],

context_summary=f"messages={len(session.messages)}",

)

except Exception: # noqa: BLE001 — never block error path on logging

logger.debug("failure_record_skipped", exc_info=True)

# ainput leaves stdout non-blocking; restore before large writes.

restore_blocking_stdio()

await aprint(f"Assistant: {result.message}", flush=True)

restore_blocking_stdio()

await aprint(f"Assistant: {result.message}", flush=True)

from __future__ import annotations

import json

import logging

import re

from datetime import UTC, datetime

from pathlib import Path

from ai_agent.application.agent import Agent

from ai_agent.domain.models import StepResult

from ai_agent.domain.ports import ApprovalGate

from ai_agent.domain.state import ConversationState

logger = logging.getLogger(__name__)

BRIEF_PROMPT_TEMPLATE = """Produce a research brief on the following topic.

def slugify(topic: str, *, max_len: int = 60) -> str:

"""Filesystem-safe slug from a research topic."""

cleaned = re.sub(r"[^a-zA-Z0-9]+", "-", topic.strip().lower()).strip("-")

if not cleaned:

cleaned = "brief"

return cleaned[:max_len].rstrip("-")

def render_brief_markdown(

topic: str,

result: StepResult,

*,

include_tool_traces: bool = True,

) -> str:

"""Compose the on-disk brief from the agent step result."""

stamp = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")

parts = [

f"# Research brief: {topic}",

"",

f"_Generated {stamp} · rounds={result.rounds_used} · kind={result.kind}_",

"",

result.message.strip() or "(empty response)",

"",

]

if include_tool_traces and result.tool_results:

parts.extend(

[

"## Tool traces",

"",

"```json",

json.dumps(result.tool_results, indent=2),

"```",

"",

]

)

return "\n".join(parts)

async def run_research_brief(

topic: str,

*,

agent: Agent,

session: ConversationState | None = None,

output_dir: Path,

approval_gate: ApprovalGate | None = None,

require_approval: bool = False,

) -> Path:

"""

cleaned = topic.strip()

if not cleaned:

raise ValueError("topic must be non-empty")

session = session or agent.create_session()

# Skip deterministic greeting for one-shot briefs.

session.greeting_sent = True

prompt = BRIEF_PROMPT_TEMPLATE.format(topic=cleaned)

result = await agent.step(session=session, user_input=prompt)

if result.kind == "error":

raise RuntimeError(result.message)

if require_approval:

if approval_gate is None:

raise ValueError("require_approval=True needs an ApprovalGate")

approved = await approval_gate.request(

f"Publish research brief on {cleaned!r} to {output_dir}?"

)

if not approved:

raise PermissionError("Brief publication declined by user")

output_dir.mkdir(parents=True, exist_ok=True)

stamp = datetime.now(UTC).strftime("%Y%m%d")

path = output_dir / f"{stamp}_{slugify(cleaned)}.md"

path.write_text(render_brief_markdown(cleaned, result), encoding="utf-8")

logger.info("research_brief_written path=%s", path)

return path