This repository has no description
0

Configure Feed

Select the types of activity you want to include in your feed.

at main 3.5 kB View raw
1"""File tools for issue investigation (knot git only).""" 2 3from __future__ import annotations 4 5import json 6import os 7from typing import Any 8 9import httpx 10from langchain_core.tools import BaseTool, tool 11 12from agent.context import IssueSessionContext 13from agent.tangled_client import DEFAULT_TIMEOUT, list_tree, normalize_tree_entries, read_blob 14 15DEFAULT_MAX_FILE_CHARS = 32_000 16 17 18def _truncate(text: str, limit: int) -> dict[str, Any]: 19 if len(text) <= limit: 20 return {"content": text, "truncated": False, "size_chars": len(text)} 21 return { 22 "content": text[:limit], 23 "truncated": True, 24 "size_chars": len(text), 25 "note": f"truncated to {limit} chars; request a narrower path or smaller file", 26 } 27 28 29def make_file_tools( 30 ctx: IssueSessionContext, 31 *, 32 max_file_chars: int | None = None, 33) -> list[BaseTool]: 34 """Build tools bound to a single issue session (repo/knot from context).""" 35 limit = max_file_chars or int(os.getenv("AGENT_MAX_FILE_CHARS", str(DEFAULT_MAX_FILE_CHARS))) 36 37 @tool 38 def read_repo_file(path: str, ref: str | None = None) -> str: 39 """Read exact file contents from the issue's repository on the knot. 40 41 Args: 42 path: File path relative to repo root (e.g. README.md, src/lib.rs). 43 ref: Git ref (branch/tag/commit). Defaults to the session ref. 44 """ 45 git_ref = ref or ctx.ref 46 path = path.lstrip("/") 47 with httpx.Client(timeout=DEFAULT_TIMEOUT, follow_redirects=True) as client: 48 try: 49 text = read_blob( 50 client, 51 knot_hostname=ctx.knot_hostname, 52 repo_did=ctx.repo_did, 53 path=path, 54 ref=git_ref, 55 ) 56 except Exception as exc: # noqa: BLE001 - return to model as tool output 57 return json.dumps({"error": str(exc), "path": path, "ref": git_ref}) 58 payload = _truncate(text, limit) 59 payload.update({"path": path, "ref": git_ref, "repo_did": ctx.repo_did}) 60 return json.dumps(payload, ensure_ascii=False) 61 62 @tool 63 def list_repo_files(path: str = "", ref: str | None = None) -> str: 64 """List files in a repository directory on the knot. 65 66 Use only when the session file tree is insufficient. Prefer known paths 67 from context when possible. 68 69 Args: 70 path: Directory relative to repo root (empty string = root). 71 ref: Git ref. Defaults to the session ref. 72 """ 73 git_ref = ref or ctx.ref 74 directory = path.lstrip("/") 75 with httpx.Client(timeout=DEFAULT_TIMEOUT, follow_redirects=True) as client: 76 try: 77 tree = list_tree( 78 client, 79 knot_hostname=ctx.knot_hostname, 80 repo_did=ctx.repo_did, 81 path=directory, 82 ref=git_ref, 83 ) 84 entries = normalize_tree_entries(tree) 85 except Exception as exc: # noqa: BLE001 86 return json.dumps( 87 {"error": str(exc), "path": directory or "/", "ref": git_ref} 88 ) 89 return json.dumps( 90 { 91 "path": directory or "/", 92 "ref": git_ref, 93 "entries": entries, 94 }, 95 ensure_ascii=False, 96 ) 97 98 return [read_repo_file, list_repo_files]