HKUDS/OpenHarness — Nghiên cứu sâu các kỹ thuật Harness

Code từ OpenHarness

turn_count
      = 0 while
      context.max_turns is None or turn_count < context.max_turns: turn_count +=
      1 # --- auto-compact check
      before calling the model --------------- async
      for event, usage in
      _stream_compaction(trigger="auto"): yield event, usage messages, was_compacted =
      last_compaction_result final_message: ConversationMessage | None = None async for event in
      context.api_client.stream_message( ApiMessageRequest(model=context.model,
      messages=messages, system_prompt=context.system_prompt,
      tools=context.tool_registry.to_api_schema()): if
      isinstance(event, ApiTextDeltaEvent): yield
      AssistantTextDelta(text=event.text), None elif isinstance(event, ApiMessageCompleteEvent):
      final_message = event.message if not
      final_message.tool_uses: return # natural end of turn tool_calls =
      final_message.tool_uses if len(tool_calls) ==
      1: # Single tool: sequential
      (stream events immediately) tc = tool_calls[0] yield
      ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None result = await
      _execute_tool_call(context, tc.name, tc.id, tc.input) yield ToolExecutionCompleted(tool_name=tc.name,
      output=result.content, is_error=result.is_error), None tool_results = [result] else: # Multiple tools: execute
      concurrently, emit events after for tc
      in tool_calls: yield
      ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None # return_exceptions=True tránh
      orphan tool_use blocks raw_results = await
      asyncio.gather( *[_run(tc) for tc in tool_calls], return_exceptions=True ) # wrap exceptions thành
      ToolResultBlock(is_error=True) ...
      messages.append(ConversationMessage(role="user",
      content=tool_results))

Code example (generic)

async
      def run_agent_turn(messages, tools): resp =
      await llm.call(messages, tools) if not resp.tool_uses: return
      resp if len(resp.tool_uses) == 1: tc = resp.tool_uses[0]
      results = [await exec_tool(tc)] else: # CRITICAL:
      return_exceptions=True # otherwise failed tool
      leaves siblings cancelled # → orphan tool_use
      blocks → next API call rejected raw = await
      asyncio.gather( *[exec_tool(tc) for tc in resp.tool_uses], return_exceptions=True, ) results = [wrap_exception(tc, r) for tc, r in zip(resp.tool_uses,
      raw)] messages.append({"role": "user", "content": results})
      return results

Code từ OpenHarness

# 1. Before-turn
      auto-compact async for event, usage in _stream_compaction(trigger="auto"): yield event, usage
      messages, was_compacted = last_compaction_result # 2.
      Reactive compact khi provider trả prompt-too-long except Exception as exc:
      error_msg = str(exc) if not
      reactive_compact_attempted and
      _is_prompt_too_long_error(exc): reactive_compact_attempted = True yield
      StatusEvent(message=REACTIVE_COMPACT_STATUS_MESSAGE), None async for event, usage
      in _stream_compaction(trigger="reactive", force=True): yield event, usage messages, was_compacted =
      last_compaction_result if was_compacted: continue # retry turn với messages đã
      compact # 3. services/compact/microcompact.py —
      cheap, fast first pass # Drop stale
      ToolResultBlock content (keep tool_use structure) # 4. services/compact/summary.py — LLM summarize khi
      microcompact không đủ # 5.
      services/compact/ptl_retry.py — truncate_head_for_ptl_retry() # last-chance cắt head khi reactive vẫn
      overflow

Code example (generic)

class CompactionOrchestrator:
      async def maybe_compact(self, messages, trigger="auto"): estimate =
      self.token_counter.estimate(messages) if estimate
      < self.threshold: return messages, False # Phase 1: cheap — drop old
      ToolResultBlock content only pruned = self.microcompact(messages)
      if self.token_counter.estimate(pruned) <
      self.threshold: return pruned, True # Phase 2: expensive — LLM
      summarize summary = await
      self.llm_summarize(pruned) return [system_prompt,
      summary] + pruned[-5:], True async def run_turn(self, messages): messages, _ = await self.maybe_compact(messages) try: return await
      llm.call(messages) except PromptTooLongError:
      messages, _ = await self.maybe_compact(messages,
      trigger="reactive") return
      await llm.call(messages) # one
      retry

Code từ OpenHarness

from dataclasses
      import dataclass from typing import
      Literal, Union # 7 event types yielded from
      run_query() @dataclass(frozen=True) class AssistantTextDelta:
      text: str @dataclass(frozen=True) class AssistantTurnComplete:
      message: ConversationMessage; usage: UsageSnapshot @dataclass(frozen=True) class
      ToolExecutionStarted: tool_name: str;
      tool_input: dict @dataclass(frozen=True) class ToolExecutionCompleted:
      tool_name: str; output: object; is_error: bool @dataclass(frozen=True) class
      StatusEvent: message: str @dataclass(frozen=True) class
      ErrorEvent: message: str @dataclass(frozen=True) class
      CompactProgressEvent: phase: Literal[ "hooks_start", "context_collapse_start", "context_collapse_end", "session_memory_start", "session_memory_end", "compact_start", "compact_retry", "compact_end", "compact_failed", ] message: str StreamEvent = Union[
      AssistantTextDelta, AssistantTurnComplete, ToolExecutionStarted,
      ToolExecutionCompleted, StatusEvent, ErrorEvent, CompactProgressEvent,
      ]

Code example (generic)

from typing import Union, Literal from dataclasses import dataclass @dataclass(frozen=True) class
      TextDelta: text: str @dataclass(frozen=True) class
      ToolStart: tool: str; input: dict @dataclass(frozen=True) class
      ToolEnd: tool: str; output: str; error: bool
      Event = Union[TextDelta, ToolStart, ToolEnd] async
      def render_events(gen): async for ev in gen: match ev:
      case TextDelta(text): console.print(text,
      end="") case
      ToolStart(tool, _): ui.mark_tool(tool, "running")
      case ToolEnd(tool, _, error): ui.mark_tool(tool,
      "error" if error else "done")

Code từ OpenHarness

async def
      _execute_tool_call( context: QueryContext,
      tool_name: str, tool_use_id: str, tool_input: dict[str, object], ) ->
      ToolResultBlock: # ---- PRE_TOOL_USE hook ----
      if context.hook_executor is not
      None: pre_hooks = await
      context.hook_executor.execute( HookEvent.PRE_TOOL_USE, {"tool_name": tool_name, "tool_input": tool_input, "event": HookEvent.PRE_TOOL_USE.value}, ) if pre_hooks.blocked: return
      ToolResultBlock( tool_use_id=tool_use_id, content=pre_hooks.reason or f"pre_tool_use hook blocked {tool_name}", is_error=True, ) # ---- Actual tool execution
      ---- result = await
      context.tool_registry.execute(tool_name, tool_input, ...) # ---- POST_TOOL_USE hook (observation only, cannot block)
      ---- if context.hook_executor is not None: await
      context.hook_executor.execute( HookEvent.POST_TOOL_USE, {"tool_name": tool_name, "tool_input": tool_input, "tool_result": result.content, "is_error": result.is_error}, ) return result

Code example (generic)

class HookExecutor: async def execute(self, event: HookEvent, payload: dict) ->
      HookResult: blocked = False; reason = None for hook in self.hooks.get(event, []): res = await hook.run(payload) if
      res.blocked: blocked = True; reason = res.reason;
      break return
      HookResult(blocked=blocked, reason=reason) # Usage
      inside tool execution async def run_tool(name, args): pre = await hooks.execute(HookEvent.PRE_TOOL_USE, {"tool_name": name, "args":
      args}) if pre.blocked: return {"error": pre.reason
      or "blocked by hook"}
      result = await tool.execute(args) await hooks.execute(HookEvent.POST_TOOL_USE, {"tool_name": name, "result":
      result}) return result

Code từ OpenHarness

from collections
      import deque @dataclass class QueryContext: # Recent work tracking — persisted across compaction
      rounds recent_goals: deque[str] = field(default_factory=lambda: deque(maxlen=5))
      recent_reads: deque[str] = field(default_factory=lambda: deque(maxlen=6))
      active_artifacts: deque[str] = field(default_factory=lambda: deque(maxlen=8))
      async_agent_tasks: deque[AgentTaskRef] = field(default_factory=lambda: deque(maxlen=12))
      work_log: deque[WorkLogEntry] = field(default_factory=lambda: deque(maxlen=10)) ...
      # Updated on every ToolExecutionCompleted: def record_tool_execution(ctx:
      QueryContext, tc: ToolUseBlock, result: ToolResultBlock): if tc.name == "Read": path =
      tc.input.get("file_path") if path: ctx.recent_reads.append(path) elif tc.name == "Write" or tc.name == "Edit": path =
      tc.input.get("file_path") if path: ctx.active_artifacts.append(path) elif tc.name == "Task":
      ctx.async_agent_tasks.append(AgentTaskRef(id=..., status="pending"))
      ctx.work_log.append(WorkLogEntry(tool=tc.name, ts=time.time(), ok=not result.is_error)) # When compact
      happens, these deques get injected into summary: # "Recently read: path1, path2, ..." # "Active artifacts: draft.md, report.html, ..." # "Pending subagents: task-abc (status=running),
      ..."

Code example (generic)

from collections import deque class AgentMemory: def __init__(self):
      self.recent_goals = deque(maxlen=5)
      self.recent_reads = deque(maxlen=6)
      self.pending_tasks = deque(maxlen=12) def on_tool(self, name, args,
      ok): if name == "Read":
      self.recent_reads.append(args["file_path"]) elif name == "Task":
      self.pending_tasks.append(args["id"]) def summary_block(self) ->
      str: return ( f"Recent files read: {', '.join(self.recent_reads)}\n"
      f"Active artifacts: {',
      '.join(self.active_artifacts)}\n" f"Pending subagents: {', '.join(self.pending_tasks)}" )
      # Inject into compact summary prompt: compact_msg
      = f"Previous conversation summary:
      ...\n\n{memory.summary_block()}"

Code từ OpenHarness

def build_system_prompt(ctx: PromptContext) -> str:
      sections: list[str] = [] # 1. Base role — persona +
      operating principles sections.append(BASE_PROMPT) # 2. Environment — cwd, platform, python, git branch,
      model sections.append(_build_env_section(ctx)) #
      3. Effort + reasoning passes (thinking mode hint) if ctx.effort:
      sections.append(_build_effort_section(ctx.effort, ctx.passes)) # 4. Skills — discovered skills injected như
      "tools-of-tools" if ctx.skills:
      sections.append(_build_skills_section(ctx.skills)) # 5.
      Delegation — khi có subagents if
      ctx.subagent_definitions:
      sections.append(_build_delegation_section(ctx.subagent_definitions)) # 6. CLAUDE.md — cascading from cwd to root if ctx.claudemd_content: sections.append(f"# Local project rules
      (CLAUDE.md)\n{ctx.claudemd_content}") # 7. Local
      rules (.claude/rules/*.md) if
      ctx.local_rules: sections.append(_build_local_rules(ctx.local_rules))
      # 8. Issue/PR context (khi chạy trong GitHub
      autopilot) if ctx.issue_or_pr:
      sections.append(_build_issue_section(ctx.issue_or_pr)) #
      9. Relevant memories (top-k from memory search) if ctx.relevant_memories:
      sections.append(_build_memories_section(ctx.relevant_memories)) return "\n\n---\n\n".join(sections)

Code example (generic)

@dataclass class PromptContext: cwd:
      Path; model: str; effort: str | None
      claudemd_content: str | None subagent_definitions:
      list[AgentDef] relevant_memories: list[MemoryHeader] def build_system_prompt(ctx:
      PromptContext) -> str: parts = [ BASE_ROLE, # 1.
      stable build_env(ctx), # 2. stable per
      session build_effort(ctx.effort), # 3. stable per
      mode build_subagents(ctx.subagent_definitions), #
      4. stable build_claudemd(ctx.claudemd_content), #
      5. changes when cwd changes build_memories(ctx.relevant_memories),
      # 6. changes every turn ] return "\n\n---\n\n".join(p
      for p in parts if p)

Code từ OpenHarness

def discover_claudemd(cwd: Path, *, max_chars: int = 12000) -> str: collected: list[tuple[Path, str]] =
      [] seen: set[Path] = set() # Walk upward from cwd to
      filesystem root current = cwd.resolve() while
      True: for candidate in ( current / "CLAUDE.md",
      current / ".claude" / "CLAUDE.md", ): if
      candidate.exists() and candidate not in seen: seen.add(candidate) content =
      candidate.read_text(encoding="utf-8",
      errors="replace") if
      len(content) > max_chars: content = content[:max_chars] + f"\n\n... [truncated at {max_chars}
      chars]" collected.append((candidate, content)) #
      Also collect .claude/rules/*.md at each level rules_dir = current /
      ".claude" / "rules"
      if rules_dir.is_dir(): for
      rule in sorted(rules_dir.glob("*.md")): if rule not in seen: seen.add(rule) collected.append((rule,
      rule.read_text(...))) if current.parent ==
      current: break # reached
      root current = current.parent # Reverse so
      root-level rules come first, leaf rules last (last wins)
      collected.reverse() return "\n\n".join(f"## {p.relative_to(cwd) if ...}\n\n{c}" for p, c in
      collected)

Code example (generic)

def find_up(cwd: Path, filenames: list[str]) ->
      list[Path]: """Walk upward from cwd to root, collecting
      matching files.""" found, seen = [], set()
      current = cwd.resolve() while True: for name in filenames: p =
      current / name if p.exists() and p not in seen: seen.add(p);
      found.append(p) if current.parent == current:
      break current = current.parent return found def assemble_rules(cwd: Path) -> str: files =
      find_up(cwd, ["CLAUDE.md", ".claude/CLAUDE.md", "AGENTS.md"]) # Root-first so leaf
      overrides (last-wins) files.reverse() return "\n\n".join(f.read_text()[:12000] for f in files)

Code từ OpenHarness

from
      hashlib import sha1 from
      pathlib import Path from
      openharness.config.paths import get_data_dir def get_project_memory_dir(cwd:
      str | Path) -> Path: """Return the persistent memory
      directory for a project.""" path = Path(cwd).resolve() digest =
      sha1(str(path).encode("utf-8")).hexdigest()[:12]
      memory_dir = get_data_dir() / "memory" / f"{path.name}-{digest}"
      memory_dir.mkdir(parents=True, exist_ok=True) return memory_dir def get_memory_entrypoint(cwd:
      str | Path) -> Path: """Return the project memory
      entrypoint file.""" return
      get_project_memory_dir(cwd) / "MEMORY.md" # Result:
      ~/.openharness/memory/{project-name}-{sha1[:12]}/ # ├── MEMORY.md ← entrypoint (index) # ├── architecture.md ← topic memory # ├── api-design.md # └──
      ...

Code example (generic)

from hashlib import sha1 from pathlib import Path def project_memory_dir(cwd:
      Path, data_root: Path) -> Path: """Stable,
      collision-resistant dir per project path.""" abs_path =
      cwd.resolve() digest = sha1(str(abs_path).encode()).hexdigest()[:12] d = data_root / "memory" /
      f"{abs_path.name}-{digest}" d.mkdir(parents=True, exist_ok=True) return d # Usage mem_dir =
      project_memory_dir(Path.cwd(), Path.home() / ".myagent") # →
      ~/.myagent/memory/app-a3f8b7c1d2e5/

Code từ OpenHarness

def find_relevant_memories(query: str, cwd, *,
      max_results=5) -> list[MemoryHeader]: tokens =
      _tokenize(query) if not tokens: return [] scored: list[tuple[float, MemoryHeader]] = []
      for header in
      scan_memory_files(cwd, max_files=100): meta =
      f"{header.title}
      {header.description}".lower() body = header.body_preview.lower()
      # Metadata matches weighted 2x; body matches 1x.
      meta_hits = sum(1 for t in tokens if t in meta) body_hits = sum(1 for t in tokens if t in body) score = meta_hits
      * 2.0 + body_hits if
      score > 0: scored.append((score, header))
      # Rank by score desc, then recency desc
      (tie-break) scored.sort(key=lambda item:
      (-item[0], -item[1].modified_at)) return [header
      for _, header in
      scored[:max_results]] def _tokenize(text: str) -> set[str]: """Extract search tokens, handling ASCII + Han
      ideographs.""" # ASCII word tokens (3+ chars) —
      filter out stopwords like "the" ascii_tokens = {t for t in re.findall(r"[A-Za-z0-9_]+", text.lower())
      if len(t) >= 3} # Han ideographs (each character carries independent
      meaning) han_chars = set(re.findall(r"[\u4e00-\u9fff\u3400-\u4dbf]", text)) return ascii_tokens | han_chars

Code example (generic)

import re def tokenize(text: str) -> set[str]: ascii_tok = {t for t
      in re.findall(r"\w+", text.lower()) if len(t)
      >= 3} han_tok = set(re.findall(r"[\u4e00-\u9fff]", text)) return ascii_tok | han_tok def
      rank_memories(query: str, memories, k=5): q_tokens = tokenize(query) scored = [] for m in memories: meta_hits =
      sum(1 for t in q_tokens if t in m.meta.lower())
      body_hits = sum(1 for t in q_tokens if t in m.body.lower()) score =
      meta_hits * 2.0 + body_hits if score: scored.append((score, m.modified_at, m))
      scored.sort(key=lambda t: (-t[0], -t[1])) return [m for _, _, m in scored[:k]]

Code từ OpenHarness

from abc
      import ABC, abstractmethod from
      typing import ClassVar from pydantic import BaseModel class BaseTool(ABC): name: ClassVar[str] description:
      ClassVar[str] input_model: ClassVar[type[BaseModel]] is_read_only:
      ClassVar[bool] = False @classmethod def to_schema(cls) -> dict: """JSON
      Schema auto-generated from Pydantic input model.""" return { "name": cls.name,
      "description": cls.description, "input_schema": cls.input_model.model_json_schema(), }
      @abstractmethod async def
      execute(self, args: BaseModel, ctx: ToolContext)
      -> ToolResult: """Execute tool with validated args;
      return result or error.""" ... # Concrete
      usage class ReadInput(BaseModel): file_path: str offset: int |
      None = None limit: int |
      None = None class ReadTool(BaseTool): name
      = "Read" description = "Read a
      file..." input_model = ReadInput is_read_only = True async def execute(self, args: ReadInput, ctx) -> ToolResult:
      ...

Code example (generic)

from pydantic import BaseModel, Field from abc
      import ABC, abstractmethod class BaseTool(ABC): name:
      str input_model: type[BaseModel] @classmethod
      def schema(cls): return {"name": cls.name, "input_schema": cls.input_model.model_json_schema()}
      @abstractmethod async def
      execute(self, args: BaseModel) -> dict: ... class GrepInput(BaseModel): pattern: str = Field(description="regex
      pattern") path: str | None = None class GrepTool(BaseTool): name
      = "Grep" input_model = GrepInput async def execute(self, args:
      GrepInput): return await
      run_ripgrep(args.pattern, args.path)

Code từ OpenHarness

#
      tools/bash_tool.py MAX_BASH_OUTPUT_BYTES = 12 * 1024 #
      12KB def _truncate_output(raw: bytes, limit: int =
      MAX_BASH_OUTPUT_BYTES) -> str: # 1. Decode UTF-8 with
      replace policy (never raise on bad bytes) text = raw.decode("utf-8", errors="replace")
      if len(text) <= limit: return text # 2. Keep head + tail,
      drop middle with clear marker half = limit // 2 return ( text[:half] + f"\n\n... [output truncated:
      {len(text) - limit} bytes omitted] ...\n\n" + text[-half:] ) # tools/web_fetch_tool.py MAX_WEB_FETCH_BYTES = 50 * 1024 #
      50KB — web pages can be larger #
      tools/file_read_tool.py — different strategy: line-based not
      byte-based DEFAULT_READ_LIMIT = 200 # lines MAX_LINE_LENGTH = 2000
      # truncate long lines def
      _read_with_limits(path: Path, offset: int, limit:
      int) -> str: lines = [] with path.open("r", encoding="utf-8",
      errors="replace") as f:
      for i, line in
      enumerate(f): if i < offset: continue if i >= offset +
      limit: break if len(line)
      > MAX_LINE_LENGTH: line = line[:MAX_LINE_LENGTH] + "... [line truncated]\n" lines.append(f"{i+1}\t{line}") return "".join(lines)

Code example (generic)

def truncate_mid(text: str,
      limit: int) -> str: if len(text) <= limit:
      return text half = limit // 2 dropped = len(text) - limit return text[:half] + f"\n... [{dropped} chars truncated] ...\n" +
      text[-half:] def safe_decode(raw: bytes) -> str: return raw.decode("utf-8",
      errors="replace") # Per-tool
      policy LIMITS = {"Bash": 12*1024, "WebFetch": 50*1024, "Read": 200} # line count for
      Read

Code từ OpenHarness

import
      pty, os, shlex, signal, asyncio INTERACTIVE_MARKERS = ( ("npm create", "--yes"), ("npx create", "--yes"), ("pnpm create", "--yes"),
      ("yarn create", "--yes"), ("bun create", "--yes"), ("pip install",
      "--quiet"), # may prompt for
      confirmation ... ) def _preflight_interactive_command(cmd: str) -> str |
      None: """Return error msg if cmd
      would prompt interactively.""" tokens = shlex.split(cmd) joined =
      " ".join(tokens).lower() for marker, required_flag in
      INTERACTIVE_MARKERS: if marker in joined and required_flag
      not in joined: return
      f"Command likely requires
      interaction. Add {required_flag} or equivalent." return None async def run_bash(cmd: str, timeout: float = 600) -> BashResult: # 1. Preflight
      check — reject scaffolds without --yes/--ci if err := _preflight_interactive_command(cmd): return BashResult(exit_code=2,
      stderr=err, is_error=True) # 2.
      Allocate PTY so that TTY-detecting programs work (vim -c, ansi
      colors) master_fd, slave_fd = pty.openpty() proc = await asyncio.create_subprocess_exec( "bash", "-c", cmd,
      stdin=slave_fd, stdout=slave_fd, stderr=slave_fd, #
      merge stderr→stdout preexec_fn=os.setsid, # own
      process group ) # 3. Read from master fd with
      overall timeout output = bytearray() try:
      await asyncio.wait_for(_read_pty(master_fd,
      output), timeout=timeout) except
      asyncio.TimeoutError: # 4. Graceful: SIGTERM → wait 2s →
      SIGKILL os.killpg(os.getpgid(proc.pid), signal.SIGTERM) try: await
      asyncio.wait_for(proc.wait(), timeout=2.0) except asyncio.TimeoutError:
      os.killpg(os.getpgid(proc.pid), signal.SIGKILL) return BashResult(exit_code=-1,
      output=bytes(output), timed_out=True) return BashResult(exit_code=proc.returncode,
      output=bytes(output))

Code example (generic)

import asyncio, shlex, os, signal, pty INTERACTIVE =
      ("npm create", "npx
      create", "yarn create") async def safe_bash(cmd: str,
      timeout=600): for m in INTERACTIVE: if m in cmd.lower() and "--yes" not in cmd: return {"error": f"'{m}' requires --yes to run
      non-interactively"} master, slave = pty.openpty() proc = await asyncio.create_subprocess_exec( "bash", "-c", cmd,
      stdin=slave, stdout=slave, stderr=slave, preexec_fn=os.setsid, ) try: await
      asyncio.wait_for(proc.wait(), timeout) except
      asyncio.TimeoutError: os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
      try: await
      asyncio.wait_for(proc.wait(), 2) except asyncio.TimeoutError:
      os.killpg(os.getpgid(proc.pid), signal.SIGKILL) return {"exit":
      proc.returncode}

Code từ OpenHarness

import
      shutil, subprocess, fnmatch from pathlib import Path RG_BIN = shutil.which("rg") # detect at import time
      async def glob_files(pattern: str, root: Path, *,
      respect_gitignore: bool = True) -> list[Path]:
      if RG_BIN: # rg --files --glob
      PATTERN — respects .gitignore by default args = [RG_BIN, "--files", "--glob", pattern,
      str(root)] if not
      respect_gitignore: args.insert(1, "--no-ignore") # Include hidden files
      if inside a git repo if (root / ".git").is_dir(): args.insert(1, "--hidden") proc = await asyncio.create_subprocess_exec( *args,
      stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL )
      stdout, _ = await proc.communicate() return [Path(line) for line
      in stdout.decode().splitlines()] # --- Python fallback --- # Respect
      .gitignore via pathspec library; else just glob results = [] for p in root.rglob("*"): if not p.is_file(): continue if
      fnmatch.fnmatch(p.name, pattern) or
      fnmatch.fnmatch(str(p), pattern): if respect_gitignore and
      _is_gitignored(p, root): continue
      results.append(p) return results async def grep_content(pattern:
      str, root: Path, *, case_insensitive=False) ->
      list[GrepMatch]: if RG_BIN: args = [RG_BIN, "--json", "--with-filename",
      pattern, str(root)] if
      case_insensitive: args.append("-i") ... else: # Python fallback with
      re.compile() regex = re.compile(pattern, re.IGNORECASE if case_insensitive else 0) ...

Code example (generic)

import shutil, asyncio from
      pathlib import Path RG = shutil.which("rg") async def fast_glob(pattern: str, root: Path): if RG: args = [RG, "--files",
      "--glob", pattern, str(root)] if (root/".git").is_dir(): args.insert(1, "--hidden") proc = await asyncio.create_subprocess_exec(*args,
      stdout=asyncio.subprocess.PIPE) out, _ = await
      proc.communicate() return [Path(l) for l in
      out.decode().splitlines()] # fallback import fnmatch return [p for p in root.rglob("*") if p.is_file() and fnmatch.fnmatch(p.name, pattern)]

Code từ OpenHarness

import
      yaml, re from pathlib import Path from dataclasses
      import dataclass FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n(.*)",
      re.DOTALL) @dataclass class Skill: name: str
      description: str path: Path body: str def load_skills(cwd: Path) -> list[Skill]: skills:
      list[Skill] = [] seen_names: set[str] = set()
      for base in
      _skill_roots(cwd): # bundled + ~/.openharness/skills +
      plugins if not base.is_dir(): continue for skill_dir in sorted(base.iterdir()): manifest = skill_dir / "SKILL.md" if not
      manifest.exists(): continue raw =
      manifest.read_text(encoding="utf-8") match =
      FRONTMATTER_RE.match(raw) if match: meta =
      yaml.safe_load(match.group(1)) or {} body = match.group(2).strip() name = meta.get("name") or skill_dir.name
      description = meta.get("description", "") else: #
      Fallback: derive from first heading + first paragraph name =
      skill_dir.name first_heading = re.search(r"^# (.+)$", raw, re.M) title =
      first_heading.group(1) if
      first_heading else name first_para =
      raw.split("\n\n")[1]
      if "\n\n" in raw else "" description = first_para[:200] body = raw # Dedup by name,
      first wins (bundled > user > plugin priority) if name in seen_names: continue seen_names.add(name)
      skills.append(Skill(name=name, description=description, path=manifest,
      body=body)) return skills

Code example (generic)

import yaml, re from pathlib
      import Path def discover_skills(roots: list[Path]): skills = []; seen =
      set() for root in roots: if not root.is_dir():
      continue for d in sorted(root.iterdir()): mf = d / "SKILL.md" if not mf.exists():
      continue raw = mf.read_text() m = re.match(r"^---\n(.*?)\n---\n(.*)", raw,
      re.DOTALL) meta, body = (yaml.safe_load(m.group(1)), m.group(2)) if m else ({}, raw) name =
      meta.get("name", d.name) if name in seen: continue # first-wins
      seen.add(name) skills.append({"name": name, "description": meta.get("description", ""), "body": body}) return
      skills

→ Phân tích sâu T14: Markdown-based skill system + frontmatter discovery

Code từ OpenHarness

from enum
      import Enum class HookEvent(Enum): SESSION_START = "session_start" SESSION_END = "session_end" PRE_COMPACT = "pre_compact" POST_COMPACT = "post_compact" PRE_TOOL_USE = "pre_tool_use" POST_TOOL_USE = "post_tool_use" class HookType(Enum): COMMAND = "command" # exec shell command, check
      exit code PROMPT = "prompt" # LLM judges payload vs policy HTTP = "http" # POST payload to
      webhook AGENT = "agent" # spawn subagent for decision class HookExecutor: async def execute(self, event:
      HookEvent, payload: dict) -> HookResult: self._maybe_reload() # hot-reload when files changed blocked, reasons =
      False, [] for hook in self._hooks_for(event): res = await hook.run(payload) reasons.append(res.message)
      if res.blocked: blocked = True if hook.type ==
      HookType.COMMAND: break #
      short-circuit for command hooks return
      HookResult(blocked=blocked, reason=";
      ".join(reasons)) class HookReloader: """Watch hook file
      mtime; reload config when modified.""" def
      maybe_reload(self): for
      path, old_mtime in list(self._mtimes.items()): new_mtime =
      path.stat().st_mtime if new_mtime != old_mtime:
      log.info("reloading hooks from %s", path)
      self._hooks = self._parse(path) self._mtimes[path] =
      new_mtime

Code example (generic)

import asyncio,
      subprocess from enum import Enum class HookEvent(Enum): PRE_TOOL = "pre_tool"; POST_TOOL = "post_tool" class CommandHook: def __init__(self, cmd): self.cmd = cmd async def run(self, payload):
      proc = await asyncio.create_subprocess_shell(
      self.cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
      stderr=asyncio.subprocess.PIPE, ) out, err = await
      proc.communicate(json.dumps(payload).encode()) return {"blocked":
      proc.returncode != 0, "reason": err.decode()} class
      HookExecutor: def __init__(self, hooks_by_event): self.hooks =
      hooks_by_event async def fire(self, event, payload): for
      h in self.hooks.get(event, []): r = await h.run(payload) if r["blocked"]: return r return {"blocked": False}

→ Phân tích sâu T15: Hook lifecycle system (6 events · 4 types · hot reload)

Code từ OpenHarness

import
      yaml from pathlib import
      Path from dataclasses import dataclass, field @dataclass class PluginManifest: name: str version: str description:
      str skills_dir: str | None = None # relative path to skills/
      hooks_file: str | None = None # relative path to
      hooks.yaml mcp_file: str | None = None # relative path to
      mcp.json commands_dir: str | None = None # relative path to
      commands/ def load_plugins(cwd: Path) -> list[Plugin]: plugins = []
      roots = [ Path.home() / ".openharness" / "plugins", cwd / ".openharness" / "plugins", ]
      for root in roots: if not root.is_dir(): continue
      for plugin_dir in
      sorted(root.iterdir()): manifest_path = plugin_dir / "plugin.yaml" if not
      manifest_path.exists(): continue data =
      yaml.safe_load(manifest_path.read_text()) manifest =
      PluginManifest(**data) # Resolve relative paths to
      absolute skills = [] if
      manifest.skills_dir: skills_path = plugin_dir / manifest.skills_dir skills
      = _load_skills_from(skills_path) hooks = [] if
      manifest.hooks_file: hooks_path = plugin_dir / manifest.hooks_file hooks =
      _parse_hooks_yaml(hooks_path) mcp_servers = [] if
      manifest.mcp_file: mcp_path = plugin_dir / manifest.mcp_file mcp_servers =
      _parse_mcp_json(mcp_path) commands = [] if
      manifest.commands_dir: # Namespace:
      "plugin:<plugin_name>:<file_stem>" commands_path =
      plugin_dir / manifest.commands_dir for cmd_file
      in sorted(commands_path.glob("*.md")): commands.append(Command( name=f"plugin:{manifest.name}:{cmd_file.stem}",
      body=cmd_file.read_text(), )) plugins.append(Plugin( manifest=manifest,
      skills=skills, hooks=hooks, mcp_servers=mcp_servers, commands=commands, ))
      return plugins

Code example (generic)

import yaml from pathlib import Path def load_plugins(roots: list[Path]) -> list: out = [] for root in roots: for d in sorted(root.glob("*")) if root.is_dir() else []: mf = d
      / "plugin.yaml" if not
      mf.exists(): continue data =
      yaml.safe_load(mf.read_text()) plugin = { "name":
      data["name"], "root": d,
      "skills": load_skills_dir(d / data.get("skills_dir", "skills")),
      "hooks": load_hooks(d / data["hooks_file"]) if
      data.get("hooks_file") else [], "mcp": load_mcp(d /
      data["mcp_file"]) if
      data.get("mcp_file") else
      [], } out.append(plugin) return out

→ Phân tích sâu T16: Plugin manifest-based loading

Code từ OpenHarness

from
      contextlib import AsyncExitStack from mcp import ClientSession,
      StdioServerParameters from mcp.client.stdio import stdio_client from
      mcp.client.sse import sse_client from pydantic import
      create_model class McpClientManager: def __init__(self, server_configs): self._configs =
      server_configs self._stack: AsyncExitStack | None
      = None self._sessions: dict[str, ClientSession] = {} self._tools: list[McpTool] = [] async def
      __aenter__(self): self._stack = AsyncExitStack()
      await self._stack.__aenter__() for name, cfg in
      self._configs.items(): if cfg.transport == "stdio": params =
      StdioServerParameters(command=cfg.command, args=cfg.args, env=cfg.env)
      stream_ctx = stdio_client(params) elif
      cfg.transport == "http": stream_ctx =
      sse_client(cfg.url, headers=cfg.headers) else:
      raise ValueError(f"Unknown MCP transport: {cfg.transport}") streams =
      await self._stack.enter_async_context(stream_ctx)
      session = ClientSession(*streams) await
      self._stack.enter_async_context(session) await
      session.initialize() self._sessions[name] = session #
      Discover tools + build Pydantic adapter for each resp = await session.list_tools() for
      tool in resp.tools: InputModel =
      _build_pydantic_from_schema(tool.inputSchema, name=tool.name)
      self._tools.append(McpTool( server=name, name=f"mcp__{name}__{tool.name}",
      # namespaced tool id
      description=tool.description, input_model=InputModel, session=session, ))
      return self async def
      __aexit__(self, *exc): if
      self._stack: await self._stack.__aexit__(*exc)
      def _build_pydantic_from_schema(schema: dict, *, name: str):
      """Convert JSON schema to a dynamic Pydantic
      model.""" fields = {} for prop, spec in schema.get("properties",
      {}).items(): py_type = _json_type_to_python(spec.get("type", "string"))
      fields[prop] = (py_type, ...) # required return create_model(f"Mcp_{name}_Input", **fields)

Code example (generic)

from contextlib import
      AsyncExitStack from pydantic import create_model class McpManager: async def __aenter__(self): self.stack = await AsyncExitStack().__aenter__() self.sessions = {}
      self.tools = [] for name, cfg in self.configs.items(): ctx = stdio_client(cfg) if cfg.transport == "stdio"
      else sse_client(cfg) streams = await self.stack.enter_async_context(ctx) sess = await
      self.stack.enter_async_context(ClientSession(*streams)) await sess.initialize() for t
      in (await
      sess.list_tools()).tools: self.tools.append({"name": f"mcp__{name}__{t.name}", "model": create_model(f"M_{t.name}", **...)}) return
      self

→ Phân tích sâu T17: MCP stdio + HTTP transport + dynamic Pydantic adapter

Code từ OpenHarness

from enum
      import Enum class PermissionMode(Enum): DEFAULT = "default" # read OK; write &
      mutation need confirm PLAN = "plan" # only read; block all mutations (even bash read-only)
      FULL_AUTO = "full_auto" #
      everything OK (except sensitive paths) # Used in
      PermissionChecker.evaluate() — see T20 if
      self._settings.mode == PermissionMode.FULL_AUTO: return PermissionDecision(allowed=True, reason="Auto mode allows all
      tools") if is_read_only: return PermissionDecision(allowed=True, reason="read-only tools are
      allowed") if self._settings.mode ==
      PermissionMode.PLAN: return PermissionDecision(
      allowed=False, reason="Plan mode
      blocks mutating tools until the user exits plan mode", ) # DEFAULT: require confirmation for mutating tools
      return PermissionDecision( allowed=False, requires_confirmation=True, reason="Mutating tools require
      user confirmation in default mode. ...", )

Code example (generic)

from
      enum import Enum class
      Mode(Enum): DEFAULT = "default"; PLAN = "plan";
      FULL_AUTO = "full_auto" def decide(mode: Mode,
      is_read_only: bool) -> str: if mode == Mode.FULL_AUTO:
      return "allow" if is_read_only: return "allow" if mode == Mode.PLAN:
      return "deny" return "confirm" # DEFAULT needs user approval

→ Phân tích sâu T18: 3-mode permission system (DEFAULT / PLAN / FULL_AUTO)

Code từ OpenHarness

# Paths that are
      always denied regardless of permission mode or user config. # These protect high-value credential and key material from
      LLM-directed access # (including via prompt
      injection). Patterns use fnmatch syntax and are matched # against the fully-resolved absolute path produced by the
      query engine. SENSITIVE_PATH_PATTERNS: tuple[str, ...] = ( # SSH keys and config "*/.ssh/*", # AWS credentials
      "*/.aws/credentials", "*/.aws/config", # GCP
      credentials "*/.config/gcloud/*", # Azure credentials "*/.azure/*", # GPG keys "*/.gnupg/*", # Docker
      credentials "*/.docker/config.json", # Kubernetes credentials "*/.kube/config", # OpenHarness own
      credential stores "*/.openharness/credentials.json", "*/.openharness/copilot_auth.json", ) def evaluate(self, tool_name, *,
      is_read_only, file_path=None, command=None): # Sensitive path check runs
      FIRST, cannot be overridden if file_path:
      for candidate in
      _policy_match_paths(file_path): for pattern in SENSITIVE_PATH_PATTERNS: if
      fnmatch.fnmatch(candidate, pattern): return
      PermissionDecision( allowed=False, reason=f"Access denied: {file_path} is a
      sensitive credential path" ) # ... rest of
      evaluation ...

Code example (generic)

import fnmatch from pathlib
      import Path SENSITIVE = ( "*/.ssh/*", "*/.aws/credentials", "*/.gnupg/*", "*/.kube/config", "*/.azure/*", "*/.docker/config.json", ) def
      is_sensitive_path(path: str) -> bool: # Match against both "dir" and "dir/" to catch directory
      roots normalized = path.rstrip("/")
      candidates = (normalized, normalized + "/") return any(fnmatch.fnmatch(c, pat) for c in candidates for pat in SENSITIVE) def check_permission(path,
      mode): if is_sensitive_path(path): return {"allowed": False, "reason": f"{path} is sensitive"} # ... other checks by mode ...

→ Phân tích sâu T19: Built-in sensitive path protection (hardcoded glob)

Code từ OpenHarness

def evaluate(self, tool_name, *, is_read_only,
      file_path=None, command=None): # Layer 1: sensitive path
      protection (see T19) if file_path: for candidate in
      _policy_match_paths(file_path): for pattern in SENSITIVE_PATH_PATTERNS: if
      fnmatch.fnmatch(candidate, pattern): return
      PermissionDecision(allowed=False, reason="sensitive") # Layer 2: explicit tool
      deny list if tool_name in self._settings.denied_tools: return PermissionDecision(allowed=False, reason=f"{tool_name} is explicitly denied") #
      Layer 3: explicit tool allow list if
      tool_name in self._settings.allowed_tools: return PermissionDecision(allowed=True, reason=f"{tool_name} is explicitly allowed") # Layer 4: path rules (glob-based) if file_path and
      self._path_rules: for candidate in _policy_match_paths(file_path): for rule in self._path_rules:
      if fnmatch.fnmatch(candidate, rule.pattern): if not rule.allow: return
      PermissionDecision(allowed=False, reason=f"deny rule: {rule.pattern}")
      # Layer 5: command deny patterns (bash-specific)
      if command: for pattern
      in getattr(self._settings, "denied_commands", []): if
      isinstance(pattern, str) and fnmatch.fnmatch(command, pattern): return PermissionDecision(allowed=False, reason=f"cmd deny: {pattern}") # Layer 6:
      fall back to mode (FULL_AUTO / PLAN / DEFAULT) if self._settings.mode == PermissionMode.FULL_AUTO:
      return PermissionDecision(allowed=True) if is_read_only: return PermissionDecision(allowed=True) if self._settings.mode ==
      PermissionMode.PLAN: return
      PermissionDecision(allowed=False, reason="plan blocks mutations") return
      PermissionDecision(allowed=False,
      requires_confirmation=True) def _policy_match_paths(file_path: str) -> tuple[str, ...]: """Return path forms that
      should participate in policy matching. Appending a trailing slash lets
      glob-style deny patterns like ``*/.ssh/*`` and ``/etc/*`` match the
      directory root itself. """ normalized = file_path.rstrip("/") if not normalized: return (file_path,) return
      (normalized, normalized + "/")

Code example (generic)

def evaluate_perm(tool, *,
      path=None, cmd=None,
      settings, mode): # Layer 1: sensitive paths
      (hardcoded) if path and is_sensitive(path): return
      deny("sensitive") # Layer 2:
      tool deny list if tool in settings.denied_tools: return
      deny("tool denied") # Layer 3:
      tool allow list if tool in settings.allowed_tools: return allow("tool allowed")
      # Layer 4: path rules if
      path and (r := match_path_rule(path,
      settings.path_rules)): if not r.allow: return deny(f"path rule: {r.pattern}") # Layer 5:
      command deny if cmd and any(fnmatch.fnmatch(cmd, p) for p in
      settings.denied_commands): return deny("command deny") # Layer 6:
      mode return mode_decision(mode,
      tool.is_read_only)

→ Phân tích sâu T20: 6-layer hierarchical permission evaluation + path normalization

Code từ OpenHarness

import
      asyncio, uuid from dataclasses import dataclass @dataclass
      class PendingApproval:
      request_id: str tool_name: str tool_input: dict future:
      asyncio.Future[PermissionResolution] class BackendHost: def __init__(self): self._pending_approvals: dict[str, PendingApproval] = {}
      self._permission_lock = asyncio.Lock() async def
      request_approval(self, tool_name, tool_input, *,
      timeout=300.0): async
      with self._permission_lock: # only one approval
      at a time request_id = str(uuid.uuid4())
      fut: asyncio.Future[PermissionResolution] =
      asyncio.get_running_loop().create_future()
      self._pending_approvals[request_id] = PendingApproval(
      request_id=request_id, tool_name=tool_name, tool_input=tool_input,
      future=fut, ) # Emit event to UI with request_id; UI
      sends back resolve_approval() later await
      self._emit(ApprovalRequestEvent( request_id=request_id,
      tool_name=tool_name, tool_input=tool_input, )) try: return await
      asyncio.wait_for(fut, timeout=timeout) except
      asyncio.TimeoutError: return
      PermissionResolution(approved=False, reason="timeout") finally:
      self._pending_approvals.pop(request_id, None)
      def resolve_approval(self,
      request_id: str, approved: bool, remember: bool = False): # Called from UI when user
      clicks Allow/Deny pending = self._pending_approvals.get(request_id)
      if pending is not None and
      not pending.future.done():
      pending.future.set_result(PermissionResolution(approved=approved,
      remember=remember))

Code example (generic)

import asyncio, uuid class ApprovalHost: def __init__(self): self.pending = {} self.lock =
      asyncio.Lock() async def ask(self, payload, timeout=300)
      -> bool: async with
      self.lock: rid = str(uuid.uuid4()) fut =
      asyncio.get_running_loop().create_future() self.pending[rid] = fut await self.notify_ui({"id":
      rid, "payload": payload}) try: return await
      asyncio.wait_for(fut, timeout) except
      asyncio.TimeoutError: return False finally: self.pending.pop(rid, None) def resolve(self, rid: str,
      approved: bool): if (fut
      := self.pending.get(rid)) and not fut.done():
      fut.set_result(approved)

→ Phân tích sâu T21: Async interactive approval với UUID + 300s timeout + lock

Code từ OpenHarness

import
      asyncio, os, sys from pathlib import Path class SubprocessAgent: def __init__(self, agent_id: str, definition:
      AgentDefinition, team_dir: Path, worktree: Path | None): self.agent_id = agent_id self.definition =
      definition self.team_dir = team_dir # shared team
      state self.worktree = worktree # git worktree for
      isolation self.proc: asyncio.subprocess.Process | None = None async def spawn(self): # Build CLI args — forward relevant flags from parent
      cmd = [ sys.executable, "-m", "openharness", "run-agent",
      "--agent-id", self.agent_id, "--team-dir", str(self.team_dir), "--model",
      self.definition.model, "--effort",
      self.definition.effort, "--permission-mode",
      self.definition.permission_mode.value, ] if
      self.definition.max_turns is not None: cmd +=
      ["--max-turns", str(self.definition.max_turns)] cwd = str(self.worktree) if
      self.worktree else None #
      Inherit env but add OPENHARNESS_SWARM_AGENT_ID env =
      os.environ.copy() env["OPENHARNESS_SWARM_AGENT_ID"] = self.agent_id env["OPENHARNESS_SWARM_TEAM_DIR"] = str(self.team_dir) self.proc = await asyncio.create_subprocess_exec( *cmd, cwd=cwd,
      env=env, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE,
      stderr=asyncio.subprocess.PIPE, ) log.info("spawned
      subagent pid=%s id=%s", self.proc.pid, self.agent_id) class BackgroundTaskManager:
      """Registry of running subagents; support
      list/kill/wait.""" def __init__(self): self._agents: dict[str, SubprocessAgent]
      = {} async def create_agent_task(self, definition: AgentDefinition, *,
      worktree=None) -> str: agent_id = f"task-{uuid.uuid4().hex[:8]}"
      agent = SubprocessAgent(agent_id, definition, self.team_dir, worktree)
      await agent.spawn() self._agents[agent_id] = agent
      return agent_id

Code example (generic)

import asyncio, os, sys, uuid class SubprocessAgent: def __init__(self, agent_id,
      model, team_dir, cwd=None): self.agent_id =
      agent_id; self.model = model self.team_dir = team_dir; self.cwd = cwd
      async def spawn(self): cmd
      = [sys.executable, "-m", "myagent", "worker", "--id", self.agent_id, "--model", self.model, "--team-dir", self.team_dir] env = {**os.environ, "AGENT_ID": self.agent_id} self.proc = await asyncio.create_subprocess_exec( *cmd,
      cwd=self.cwd, env=env, stdin=asyncio.subprocess.DEVNULL,
      stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) return self.proc.pid

Code từ OpenHarness

import
      json, os, uuid, time from pathlib import Path class Mailbox: """File-based inbox for a
      single agent. Layout: inbox/ <sort_key>-<uuid>.tmp ← being
      written (atomic) <sort_key>-<uuid>.json ← delivered
      (consumable) """ def __init__(self, agent_dir: Path): self.inbox = agent_dir
      / "inbox" self.inbox.mkdir(parents=True, exist_ok=True) def send(self, envelope: dict):
      """Atomic write: write .tmp then rename to
      .json.""" sort_key = f"{time.time_ns():020d}" # nanosecond
      for ordering msg_id = uuid.uuid4().hex[:8]
      base = self.inbox / f"{sort_key}-{msg_id}" tmp = base.with_suffix(".tmp") final = base.with_suffix(".json") tmp.write_text(json.dumps(envelope,
      ensure_ascii=False)) os.rename(tmp, final) # atomic on POSIX async def
      poll(self, interval=0.25): """Yield envelopes as they
      arrive. Caller deletes after consume.""" while
      True: for msg_path in sorted(self.inbox.glob("*.json")): try: envelope =
      json.loads(msg_path.read_text()) except
      json.JSONDecodeError: continue #
      partial write? skip yield envelope
      msg_path.unlink(missing_ok=True) await asyncio.sleep(interval) #
      Envelope types (message kinds) # {"kind":
      "user_message", "text": "..."} # {"kind":
      "permission_request", "request_id": "...", "tool": "...", "args":
      {...}} # {"kind": "shutdown", "reason":
      "..."} # {"kind": "tool_result", "tool": "...",
      "result": ..., "is_error": bool}

Code example (generic)

import os, json, uuid, time, asyncio
      from pathlib import Path
      class FileMailbox: def __init__(self, inbox: Path):
      self.inbox = inbox; inbox.mkdir(parents=True,
      exist_ok=True) def send(self, msg: dict): sort_key = f"{time.time_ns():020d}" mid =
      uuid.uuid4().hex[:8] tmp = self.inbox / f"{sort_key}-{mid}.tmp" final =
      self.inbox / f"{sort_key}-{mid}.json" tmp.write_text(json.dumps(msg))
      os.rename(tmp, final) # atomic async def receive(self): while True: for p in sorted(self.inbox.glob("*.json")): msg = json.loads(p.read_text()) p.unlink()
      yield msg await
      asyncio.sleep(0.25)

Code từ OpenHarness

import
      json, time from pathlib import Path class PermissionSync: """Worker writes
      pending request; Leader resolves by writing resolution.""" def __init__(self, team_dir:
      Path, agent_id: str): self.pending_dir = team_dir / "permissions" / "pending"
      self.resolved_dir = team_dir / "permissions" /
      "resolved" self.pending_dir.mkdir(parents=True, exist_ok=True)
      self.resolved_dir.mkdir(parents=True,
      exist_ok=True) self.agent_id = agent_id async def worker_request(self,
      tool: str, tool_input: dict, *, timeout=300):
      """Called by worker: write pending request; wait for
      resolution file.""" request_id = uuid.uuid4().hex # Fast-path: read-only tools auto-approved without leader
      involvement if
      _is_read_only_heuristic(tool, tool_input): return
      PermissionResolution(approved=True, reason="read-only") pending_file = self.pending_dir / f"{request_id}.json"
      resolved_file = self.resolved_dir / f"{request_id}.json"
      pending_file.write_text(json.dumps({ "request_id": request_id, "agent_id": self.agent_id, "tool": tool, "tool_input":
      tool_input, "ts": time.time(), })) # Also send notification via mailbox for low-latency
      wake-up await
      self._notify_leader_via_mailbox(request_id) # Poll for
      resolution deadline = time.time() + timeout while time.time() < deadline: if resolved_file.exists(): data =
      json.loads(resolved_file.read_text()) pending_file.unlink(missing_ok=True) resolved_file.unlink(missing_ok=True) return
      PermissionResolution(**data) await
      asyncio.sleep(0.25) return PermissionResolution(approved=False, reason="timeout") def leader_resolve(self,
      request_id: str, approved: bool, *, reason=""):
      """Called by leader (main UI) when user
      approves/denies.""" resolved_file = self.resolved_dir / f"{request_id}.json" tmp =
      resolved_file.with_suffix(".tmp")
      tmp.write_text(json.dumps({"approved": approved,
      "reason": reason})) os.rename(tmp,
      resolved_file)

Code example (generic)

import os, json, time, asyncio, uuid class PermSync: def __init__(self, pending_dir,
      resolved_dir): self.p = pending_dir; self.r = resolved_dir async def worker_wait(self,
      tool, args, timeout=300): rid = uuid.uuid4().hex
      (self.p / f"{rid}.json").write_text(json.dumps({"tool": tool, "args": args}))
      resolved = self.r / f"{rid}.json" deadline = time.time() + timeout while time.time() < deadline: if resolved.exists(): data =
      json.loads(resolved.read_text()) resolved.unlink() return data["approved"] await asyncio.sleep(0.25) return False def leader_resolve(self, rid, approved): tmp = self.r /
      f"{rid}.tmp"
      tmp.write_text(json.dumps({"approved":
      approved})) os.rename(tmp, self.r / f"{rid}.json")

Code từ OpenHarness

import
      asyncio, re from pathlib import Path SLUG_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
      MAX_SLUG_LEN = 64 def
      validate_slug(slug: str) -> str: """Reject .., absolute paths, and control chars."""
      if not slug or len(slug)
      > MAX_SLUG_LEN: raise ValueError(f"slug length must be
      1..{MAX_SLUG_LEN}") if not
      SLUG_RE.match(slug): raise ValueError(f"invalid slug: {slug!r}")
      if ".." in slug or
      slug.startswith(("/", "-")): raise ValueError(f"slug must not traverse or start with
      /-: {slug!r}") return slug async def create_worktree(repo_root: Path, slug: str, *, base_ref:
      str = "HEAD") -> Path: """Create a shallow worktree for subagent isolation."""
      slug = validate_slug(slug) worktree_path = repo_root / ".openharness" / "worktrees" /
      slug branch = f"openharness/swarm/{slug}" # Reuse
      existing worktree if slug already exists if
      worktree_path.exists(): return worktree_path
      worktree_path.parent.mkdir(parents=True,
      exist_ok=True) cmd = ["git", "-C", str(repo_root), "worktree",
      "add", "--quiet", "-b", branch, str(worktree_path), base_ref] proc = await asyncio.create_subprocess_exec( *cmd,
      stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) _,
      stderr = await proc.communicate() if proc.returncode != 0: raise RuntimeError(f"git worktree add failed: {stderr.decode()}") return worktree_path async def
      remove_worktree(repo_root: Path, slug: str, *,
      force: bool = False): """Remove
      a worktree; orphan branch left as cleanup-later.""" slug =
      validate_slug(slug) worktree_path = repo_root / ".openharness" / "worktrees" /
      slug args = ["git", "-C", str(repo_root), "worktree", "remove", str(worktree_path)] if force:
      args.append("--force") ...

Code example (generic)

import asyncio, re from pathlib
      import Path SLUG_RE = re.compile(r"^[a-zA-Z0-9._-]{1,64}$")
      async def make_worktree(repo: Path, slug: str, base="HEAD") -> Path: if not
      SLUG_RE.match(slug) or ".." in slug: raise ValueError("bad slug") wt
      = repo / ".worktrees" / slug if wt.exists(): return wt branch
      = f"agent/{slug}" proc =
      await asyncio.create_subprocess_exec( "git", "-C", str(repo), "worktree", "add", "-b", branch, str(wt), base, stdout=asyncio.subprocess.DEVNULL,
      stderr=asyncio.subprocess.PIPE, ) _, err = await
      proc.communicate() if proc.returncode != 0: raise
      RuntimeError(err.decode()) return wt

Code từ OpenHarness

import
      yaml from pathlib import
      Path from pydantic import
      BaseModel from openharness.permissions.modes import PermissionMode class
      AgentDefinition(BaseModel): name: str
      description: str system_prompt: str tools: list[str] = [] # allowlist (empty = all) disallowed_tools: list[str] =
      [] # denylist model: str = "claude-sonnet-4-6" effort: str = "medium" # low/medium/high
      permission_mode: PermissionMode = PermissionMode.DEFAULT max_turns: int |
      None = 50 color: str =
      "cyan" isolation: str = "none" # "none" | "worktree"
      def load_agent_definitions(cwd: Path) -> dict[str,
      AgentDefinition]: defs = {} for location in [ cwd / ".openharness" /
      "agents", Path.home() / ".openharness" / "agents", ]:
      if not location.is_dir(): continue for f in sorted(location.glob("*.yaml")): data = yaml.safe_load(f.read_text()) for entry in (data if isinstance(data, list) else
      [data]): defn = AgentDefinition(**entry) defs[defn.name] = defn return defs # Example agent definition
      YAML: # name: doc-writer # description: Writes technical documentation # system_prompt: | # You are a doc
      writer. Focus on clarity, accuracy, and structure. # tools: [Read, Write, Grep, Glob] #
      permission_mode: plan # cannot mutate #
      max_turns: 30 # isolation: worktree # get own
      worktree class Coordinator: """Central router:
      decide which agent handles an incoming task.""" def __init__(self, definitions:
      dict[str, AgentDefinition]): self.definitions = definitions self.manager =
      BackgroundTaskManager() async def dispatch(self, task_description: str, preferred_agent:
      str | None = None): defn =
      self.definitions.get(preferred_agent) or
      self._pick_agent(task_description) worktree = await create_worktree(...) if
      defn.isolation == "worktree" else None return await
      self.manager.create_agent_task(defn, worktree=worktree)

Code example (generic)

import yaml from pathlib import Path from pydantic import BaseModel
      class AgentDef(BaseModel): name: str; description: str;
      system_prompt: str tools: list[str] = [] model: str = "claude-sonnet-4-6" max_turns: int = 50 isolation: str = "none"
      def load_defs(dir: Path)
      -> dict[str, AgentDef]: out = {} for f in sorted(dir.glob("*.yaml")): d = yaml.safe_load(f.read_text()) for entry in (d if isinstance(d, list) else
      [d]): defn = AgentDef(**entry) out[defn.name] = defn return out class Coordinator: async def dispatch(self, task: str, agent_name: str): defn =
      self.defs[agent_name] wt = await
      make_worktree(repo, agent_name) if defn.isolation
      == "worktree" else None
      return await self.spawn(defn, wt,
      task)

Code từ OpenHarness

import
      asyncio from dataclasses import dataclass @dataclass
      class ChannelMessage:
      channel: str # "slack" | "feishu" | "discord" |
      ... source_id: str # channel-specific thread /
      room user: str text: str raw: dict # adapter
      payload gốc class MessageBus: """Single agent, many
      channels. Inbound từ N channel → 1 queue, outbound từ agent → fan-out về
      đúng channel qua source_id.""" def __init__(self): self.inbound:
      asyncio.Queue[ChannelMessage] = asyncio.Queue(maxsize=1024) self.outbound:
      asyncio.Queue[tuple[str, str, str]] = asyncio.Queue() self._channels:
      dict[str, ChannelAdapter] = {} def register(self, name: str, adapter: ChannelAdapter):
      self._channels[name] = adapter async def start(self): tasks = [asyncio.create_task(c.run(self))
      for c in
      self._channels.values()]
      tasks.append(asyncio.create_task(self._outbound_dispatcher())) await asyncio.gather(*tasks, return_exceptions=True) async def _outbound_dispatcher(self): while True: channel, source_id,
      text = await self.outbound.get() adapter =
      self._channels.get(channel) if adapter: await adapter.send(source_id, text) #
      Feishu adapter — WebSocket long-connection với payload 40KB class FeishuAdapter(ChannelAdapter): MAX_PAYLOAD = 40_000
      # ~40KB per interactive card async def run(self, bus:
      MessageBus): client = lark.ws.Client(app_id=..., app_secret=...,
      event_handler=lambda evt: self._on_event(evt,
      bus)) await client.start() 

Code example (generic)

import asyncio from abc import ABC,
      abstractmethod class ChannelAdapter(ABC): @abstractmethod async def
      run(self, bus): ... @abstractmethod async def
      send(self, source_id: str, text: str): ... class SlackAdapter(ChannelAdapter): def __init__(self, token):
      self.app = AsyncApp(token=token) self.app.event("message")(self._on_message) async
      def _on_message(self, event, say, bus):
      await bus.inbound.put(ChannelMessage(
      channel="slack", source_id=event["channel"], user=event["user"], text=event["text"],
      raw=event, )) async def send(self, source_id, text): await
      self.app.client.chat_postMessage(channel=source_id, text=text) # Dùng một agent cho Slack + Discord đồng thời async def main(): bus =
      MessageBus() bus.register("slack",
      SlackAdapter(token="xoxb-..."))
      bus.register("discord",
      DiscordAdapter(token="...")) async def agent_loop(): while True: msg = await bus.inbound.get() reply = await llm_reply(msg.text, user=msg.user) await bus.outbound.put((msg.channel, msg.source_id,
      reply)) await asyncio.gather(bus.start(),
      agent_loop()) 

Code từ OpenHarness

import ast
      from pathlib import Path
      def list_document_symbols(file_path: Path) ->
      list[Symbol]: """Trả về top-level functions, classes,
      methods trong 1 file Python qua ast.parse() — không cần spawn language
      server ngoài.""" source = file_path.read_text(encoding="utf-8", errors="replace")
      try: tree = ast.parse(source) except SyntaxError: return []
      symbols = [] for node in
      ast.walk(tree): if isinstance(node,
      (ast.FunctionDef, ast.AsyncFunctionDef)): symbols.append(Symbol(
      name=node.name, kind="function",
      line=node.lineno, col=node.col_offset, )) elif
      isinstance(node, ast.ClassDef): symbols.append(Symbol( name=node.name,
      kind="class", line=node.lineno,
      col=node.col_offset, )) for item in node.body: if
      isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
      symbols.append(Symbol( name=f"{node.name}.{item.name}", kind="method", line=item.lineno, col=item.col_offset, ))
      return symbols def workspace_symbol_search(root: Path, query: str) ->
      list[Symbol]: """Search symbol name across all *.py
      files — case-insensitive substring.""" results = [] for py_file in root.rglob("*.py"): if
      any(part.startswith(".") for part in py_file.parts):
      continue for sym in list_document_symbols(py_file): if query.lower() in
      sym.name.lower(): results.append(sym._replace(file=py_file)) return results 

Code example (generic)

import ast from pathlib import Path def find_function_definition(file: Path, name: str) ->
      tuple[int, int] | None:
      try: tree = ast.parse(file.read_text()) except SyntaxError: return None for node in ast.walk(tree): if
      isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) \ and node.name == name: return
      (node.lineno, node.col_offset) return None def find_references(root: Path, name: str) -> list[tuple[Path, int]]: """Approximate:
      tìm Name/Attribute có id trùng.""" hits = [] for py in root.rglob("*.py"): try: tree =
      ast.parse(py.read_text()) except SyntaxError:
      continue for node in ast.walk(tree): if
      isinstance(node, ast.Name) and node.id == name:
      hits.append((py, node.lineno)) elif
      isinstance(node, ast.Attribute) and node.attr ==
      name: hits.append((py, node.lineno)) return hits
      

Code từ OpenHarness

import
      docker from docker.errors import DockerException class
      DockerSandbox: """Wrap tool
      execution trong container isolated. Tool nhận được stdin/stdout qua
      container exec, không thấy host filesystem trừ volume mount có kiểm
      soát.""" def __init__(self, session_id: str, workdir: Path):
      self.name = f"openharness-sandbox-{session_id}"
      self.workdir = workdir self._client = None
      self._container = None def
      check_available(self) -> bool: """Pre-flight: Docker daemon running? Platform
      supported?""" try: self._client =
      docker.from_env(timeout=5) self._client.ping() return True except DockerException as e:
      log.warning(f"Docker not available: {e}") return False async def start(self):
      self._container = self._client.containers.run( image="openharness/sandbox:latest", name=self.name,
      detach=True, stdin_open=True, tty=True,
      network_mode="bridge", # có
      net; dùng "none" nếu muốn offline mem_limit="2g", cpu_quota=100_000, # 1
      CPU read_only=False, #
      agent cần ghi vào /workspace volumes={ str(self.workdir): {"bind": "/workspace", "mode": "rw"}, },
      working_dir="/workspace", environment={"OPENHARNESS_SANDBOX": "1"}, )
      async def exec(self, cmd:
      list[str], timeout: int = 600) -> ExecResult: if self._container is None: raise RuntimeError("sandbox not started") result =
      self._container.exec_run( cmd=cmd, stdout=True,
      stderr=True, demux=True, )
      return ExecResult(exit_code=result.exit_code,
      output=result.output) async def cleanup(self): if
      self._container: self._container.remove(force=True) 

Code example (generic)

import docker, tempfile, subprocess
      def run_tool_in_container(cmd: list[str], workdir: str,
      timeout: int = 60): client = docker.from_env() try: container = client.containers.run( "python:3.12-slim", cmd, volumes={workdir: {"bind": "/work", "mode": "rw"}},
      working_dir="/work", mem_limit="512m", network_disabled=True,
      # no egress remove=True,
      # auto-cleanup detach=False, stdout=True, stderr=True, ) return
      container.decode("utf-8", errors="replace") except
      docker.errors.ContainerError as e: return f"[exit {e.exit_status}]:
      {e.stderr.decode()}" # Fallback if Docker
      missing def run_tool(cmd: list[str], workdir: str): if docker_available(): return
      run_tool_in_container(cmd, workdir) return
      subprocess.run(cmd, cwd=workdir, capture_output=True, text=True) 

Code từ OpenHarness

import
      asyncio, json from datetime import datetime from croniter
      import croniter @dataclass class CronJob: job_id: str name: str cron_expr: str # "*/30 * * * *" prompt: str # agent
      prompt khi fire enabled: bool = True
      last_run: str | None = None next_run: str | None =
      None class CronScheduler: TICK = 30 #
      seconds def __init__(self, registry_path: Path, history_path: Path):
      self.registry_path = registry_path self.history_path = history_path
      self.jobs: dict[str, CronJob] = self._load() def
      _load(self) -> dict[str, CronJob]: if not
      self.registry_path.exists(): return {} data =
      json.loads(self.registry_path.read_text()) return
      {j["job_id"]: CronJob(**j) for j in data} def _save(self):
      self.registry_path.write_text(json.dumps( [asdict(j) for j in self.jobs.values()],
      indent=2, )) async def run_forever(self, on_fire: Callable[[CronJob],
      Awaitable]): while True:
      now = datetime.utcnow() for job in list(self.jobs.values()): if
      not job.enabled: continue
      base = datetime.fromisoformat(job.last_run) if
      job.last_run else now itr =
      croniter(job.cron_expr, base) if
      itr.get_next(datetime) <= now: asyncio.create_task(self._fire(job,
      on_fire)) job.last_run = now.isoformat() self._save() await asyncio.sleep(self.TICK) async
      def _fire(self, job: CronJob, on_fire):
      try: result = await
      on_fire(job) self._append_history({"job_id":
      job.job_id, "fired_at":
      datetime.utcnow().isoformat(), "ok": True, "result_preview":
      result[:500]}) except Exception as e: self._append_history({"job_id": job.job_id, "ok":
      False, "error": str(e)})
      def _append_history(self,
      entry: dict): with self.history_path.open("a") as f:
      f.write(json.dumps(entry) + "\n") 

Code example (generic)

import asyncio from apscheduler.schedulers.asyncio import AsyncIOScheduler from
      apscheduler.triggers.cron import CronTrigger async def agent_job(prompt:
      str): response = await run_agent(prompt)
      save_history(prompt, response) async def main(): scheduler = AsyncIOScheduler() # Cron: Mon-Fri 09:00 scheduler.add_job( agent_job,
      trigger=CronTrigger.from_crontab("0 9 * * 1-5"),
      args=["Summarize overnight alerts"], id="daily-triage", replace_existing=True, ) scheduler.start() await
      asyncio.Event().wait() # run forever