forked from Rockachopa/Timmy-time-dashboard
Merge pull request 'feat: add Gitea issue creation — Timmy's self-improvement channel' (#9) from claude/sharp-mcnulty into main
Reviewed-on: http://localhost:3000/rockachopa/Timmy-time-dashboard/pulls/9
This commit is contained in:
rockachopa
@@ -215,6 +215,16 @@ class Settings(BaseSettings):
|
||||
thinking_enabled: bool = True
|
||||
thinking_interval_seconds: int = 300 # 5 minutes between thoughts
|
||||
thinking_distill_every: int = 10 # distill facts from thoughts every Nth thought
|
||||
thinking_issue_every: int = 20 # file Gitea issues from thoughts every Nth thought
|
||||
|
||||
# ── Gitea Integration ─────────────────────────────────────────────
|
||||
# Local Gitea instance for issue tracking and self-improvement.
|
||||
# Timmy can file issues when he notices bugs or improvement opportunities.
|
||||
gitea_url: str = "http://localhost:3000"
|
||||
gitea_token: str = "" # GITEA_TOKEN env var; falls back to ~/.config/gitea/token
|
||||
gitea_repo: str = "rockachopa/Timmy-time-dashboard" # owner/repo
|
||||
gitea_enabled: bool = True
|
||||
gitea_timeout: int = 30
|
||||
|
||||
# ── Loop QA (Self-Testing) ─────────────────────────────────────────
|
||||
# Self-test orchestrator that probes capabilities alongside the thinking loop.
|
||||
|
||||
290
src/infrastructure/hands/gitea.py
Normal file
290
src/infrastructure/hands/gitea.py
Normal file
@@ -0,0 +1,290 @@
|
||||
"""Gitea Hand — issue tracking and self-improvement channel for Timmy.
|
||||
|
||||
Provides Gitea API capabilities with:
|
||||
- Token auth (env var, config, or ~/.config/gitea/token fallback)
|
||||
- Structured result parsing
|
||||
- Dedup checks before creating issues
|
||||
- Graceful degradation: log warning, return fallback, never crash
|
||||
|
||||
Follows project conventions:
|
||||
- Config via ``from config import settings``
|
||||
- Singleton pattern for module-level import
|
||||
- Async httpx client (like Paperclip client pattern)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from difflib import SequenceMatcher
|
||||
from pathlib import Path
|
||||
|
||||
from config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_TOKEN_FILE = Path.home() / ".config" / "gitea" / "token"
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaResult:
|
||||
"""Result from a Gitea API operation."""
|
||||
|
||||
operation: str
|
||||
success: bool
|
||||
data: dict = field(default_factory=dict)
|
||||
error: str = ""
|
||||
latency_ms: float = 0.0
|
||||
|
||||
|
||||
def _resolve_token() -> str:
|
||||
"""Resolve Gitea API token from settings or filesystem fallback."""
|
||||
if settings.gitea_token:
|
||||
return settings.gitea_token
|
||||
try:
|
||||
return _TOKEN_FILE.read_text().strip()
|
||||
except (FileNotFoundError, PermissionError):
|
||||
return ""
|
||||
|
||||
|
||||
def _title_similar(a: str, b: str, threshold: float = 0.6) -> bool:
|
||||
"""Check if two issue titles are similar enough to be duplicates."""
|
||||
return SequenceMatcher(None, a.lower(), b.lower()).ratio() > threshold
|
||||
|
||||
|
||||
class GiteaHand:
|
||||
"""Gitea API hand for Timmy.
|
||||
|
||||
All methods degrade gracefully — if Gitea is unreachable or the
|
||||
API call fails, the hand returns a ``GiteaResult(success=False)``
|
||||
rather than raising.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str | None = None,
|
||||
token: str | None = None,
|
||||
repo: str | None = None,
|
||||
timeout: int | None = None,
|
||||
) -> None:
|
||||
self._base_url = (base_url or settings.gitea_url).rstrip("/")
|
||||
self._token = token or _resolve_token()
|
||||
self._repo = repo or settings.gitea_repo
|
||||
self._timeout = timeout or settings.gitea_timeout
|
||||
self._client = None
|
||||
|
||||
if not self._token:
|
||||
logger.warning(
|
||||
"Gitea token not configured — set GITEA_TOKEN or place token in %s",
|
||||
_TOKEN_FILE,
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
"GiteaHand initialised — %s/%s",
|
||||
self._base_url,
|
||||
self._repo,
|
||||
)
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Check if Gitea integration is configured and enabled."""
|
||||
return bool(settings.gitea_enabled and self._token and self._repo)
|
||||
|
||||
def _get_client(self):
|
||||
"""Lazy-initialise the async HTTP client."""
|
||||
import httpx
|
||||
|
||||
if self._client is None or self._client.is_closed:
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=self._base_url,
|
||||
headers={
|
||||
"Authorization": f"token {self._token}",
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=self._timeout,
|
||||
)
|
||||
return self._client
|
||||
|
||||
async def _request(self, method: str, path: str, **kwargs) -> GiteaResult:
|
||||
"""Make an API request with full error handling."""
|
||||
start = time.time()
|
||||
operation = f"{method.upper()} {path}"
|
||||
|
||||
if not self.available:
|
||||
return GiteaResult(
|
||||
operation=operation,
|
||||
success=False,
|
||||
error="Gitea not configured (missing token or repo)",
|
||||
)
|
||||
|
||||
try:
|
||||
client = self._get_client()
|
||||
resp = await client.request(method, path, **kwargs)
|
||||
latency = (time.time() - start) * 1000
|
||||
|
||||
if resp.status_code >= 400:
|
||||
error_body = resp.text[:500]
|
||||
logger.warning(
|
||||
"Gitea API %s returned %d: %s",
|
||||
operation,
|
||||
resp.status_code,
|
||||
error_body,
|
||||
)
|
||||
return GiteaResult(
|
||||
operation=operation,
|
||||
success=False,
|
||||
error=f"HTTP {resp.status_code}: {error_body}",
|
||||
latency_ms=latency,
|
||||
)
|
||||
|
||||
return GiteaResult(
|
||||
operation=operation,
|
||||
success=True,
|
||||
data=resp.json() if resp.text else {},
|
||||
latency_ms=latency,
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
latency = (time.time() - start) * 1000
|
||||
logger.warning("Gitea API %s failed: %s", operation, exc)
|
||||
return GiteaResult(
|
||||
operation=operation,
|
||||
success=False,
|
||||
error=str(exc),
|
||||
latency_ms=latency,
|
||||
)
|
||||
|
||||
# ── Issue operations ─────────────────────────────────────────────────
|
||||
|
||||
async def create_issue(
|
||||
self,
|
||||
title: str,
|
||||
body: str = "",
|
||||
labels: list[str] | None = None,
|
||||
) -> GiteaResult:
|
||||
"""Create an issue in the configured repository.
|
||||
|
||||
Args:
|
||||
title: Issue title (required).
|
||||
body: Issue body in markdown.
|
||||
labels: Optional list of label names (must exist in repo).
|
||||
|
||||
Returns:
|
||||
GiteaResult with issue data (number, html_url, etc.).
|
||||
"""
|
||||
owner, repo = self._repo.split("/", 1)
|
||||
payload: dict = {"title": title, "body": body}
|
||||
|
||||
# Resolve label names to IDs if provided
|
||||
if labels:
|
||||
label_ids = await self._resolve_label_ids(owner, repo, labels)
|
||||
if label_ids:
|
||||
payload["labels"] = label_ids
|
||||
|
||||
return await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{owner}/{repo}/issues",
|
||||
json=payload,
|
||||
)
|
||||
|
||||
async def list_issues(
|
||||
self,
|
||||
state: str = "open",
|
||||
labels: list[str] | None = None,
|
||||
limit: int = 50,
|
||||
) -> GiteaResult:
|
||||
"""List issues in the configured repository.
|
||||
|
||||
Args:
|
||||
state: Filter by state ("open", "closed", "all").
|
||||
labels: Filter by label names.
|
||||
limit: Max issues to return.
|
||||
|
||||
Returns:
|
||||
GiteaResult with list of issue dicts.
|
||||
"""
|
||||
owner, repo = self._repo.split("/", 1)
|
||||
params: dict = {"state": state, "limit": limit, "type": "issues"}
|
||||
if labels:
|
||||
params["labels"] = ",".join(labels)
|
||||
|
||||
return await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{owner}/{repo}/issues",
|
||||
params=params,
|
||||
)
|
||||
|
||||
async def get_issue(self, number: int) -> GiteaResult:
|
||||
"""Get a single issue by number."""
|
||||
owner, repo = self._repo.split("/", 1)
|
||||
return await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{owner}/{repo}/issues/{number}",
|
||||
)
|
||||
|
||||
async def add_comment(self, number: int, body: str) -> GiteaResult:
|
||||
"""Add a comment to an issue."""
|
||||
owner, repo = self._repo.split("/", 1)
|
||||
return await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{owner}/{repo}/issues/{number}/comments",
|
||||
json={"body": body},
|
||||
)
|
||||
|
||||
async def close_issue(self, number: int) -> GiteaResult:
|
||||
"""Close an issue."""
|
||||
owner, repo = self._repo.split("/", 1)
|
||||
return await self._request(
|
||||
"PATCH",
|
||||
f"/api/v1/repos/{owner}/{repo}/issues/{number}",
|
||||
json={"state": "closed"},
|
||||
)
|
||||
|
||||
# ── Dedup helper ─────────────────────────────────────────────────────
|
||||
|
||||
async def find_duplicate(self, title: str, threshold: float = 0.6) -> dict | None:
|
||||
"""Check if an open issue with a similar title already exists.
|
||||
|
||||
Returns the matching issue dict, or None if no duplicate found.
|
||||
"""
|
||||
result = await self.list_issues(state="open", limit=100)
|
||||
if not result.success or not isinstance(result.data, list):
|
||||
return None
|
||||
|
||||
for issue in result.data:
|
||||
existing_title = issue.get("title", "")
|
||||
if _title_similar(title, existing_title, threshold):
|
||||
return issue
|
||||
|
||||
return None
|
||||
|
||||
# ── Label helper ─────────────────────────────────────────────────────
|
||||
|
||||
async def _resolve_label_ids(self, owner: str, repo: str, label_names: list[str]) -> list[int]:
|
||||
"""Resolve label names to IDs. Returns empty list on failure."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{owner}/{repo}/labels",
|
||||
)
|
||||
if not result.success or not isinstance(result.data, list):
|
||||
return []
|
||||
|
||||
name_to_id = {label["name"].lower(): label["id"] for label in result.data}
|
||||
return [name_to_id[name.lower()] for name in label_names if name.lower() in name_to_id]
|
||||
|
||||
# ── Status ───────────────────────────────────────────────────────────
|
||||
|
||||
def info(self) -> dict:
|
||||
"""Return a status summary for the dashboard."""
|
||||
return {
|
||||
"base_url": self._base_url,
|
||||
"repo": self._repo,
|
||||
"available": self.available,
|
||||
"has_token": bool(self._token),
|
||||
}
|
||||
|
||||
|
||||
# ── Module-level singleton ──────────────────────────────────────────────────
|
||||
gitea_hand = GiteaHand()
|
||||
@@ -228,6 +228,9 @@ class ThinkingEngine:
|
||||
# Post-hook: distill facts from recent thoughts periodically
|
||||
self._maybe_distill()
|
||||
|
||||
# Post-hook: file Gitea issues for actionable observations
|
||||
self._maybe_file_issues()
|
||||
|
||||
# Post-hook: update MEMORY.md with latest reflection
|
||||
self._update_memory(thought)
|
||||
|
||||
@@ -378,6 +381,88 @@ class ThinkingEngine:
|
||||
except Exception as exc:
|
||||
logger.debug("Thought distillation skipped: %s", exc)
|
||||
|
||||
def _maybe_file_issues(self) -> None:
|
||||
"""Every N thoughts, classify recent thoughts and file Gitea issues.
|
||||
|
||||
Asks the LLM to review recent thoughts for actionable items —
|
||||
bugs, broken features, stale state, or improvement opportunities.
|
||||
Creates Gitea issues for anything worth tracking, with dedup to
|
||||
avoid flooding.
|
||||
|
||||
Only runs when:
|
||||
- Gitea is enabled and configured
|
||||
- Thought count is divisible by thinking_issue_every
|
||||
- LLM extracts at least one actionable item
|
||||
"""
|
||||
try:
|
||||
interval = settings.thinking_issue_every
|
||||
if interval <= 0:
|
||||
return
|
||||
|
||||
count = self.count_thoughts()
|
||||
if count == 0 or count % interval != 0:
|
||||
return
|
||||
|
||||
# Check Gitea availability before spending LLM tokens
|
||||
from infrastructure.hands.gitea import gitea_hand
|
||||
|
||||
if not gitea_hand.available:
|
||||
return
|
||||
|
||||
recent = self.get_recent_thoughts(limit=interval)
|
||||
if len(recent) < interval:
|
||||
return
|
||||
|
||||
thought_text = "\n".join(f"- [{t.seed_type}] {t.content}" for t in reversed(recent))
|
||||
|
||||
classify_prompt = (
|
||||
"You are reviewing your own recent thoughts for actionable items.\n"
|
||||
"Extract 0-2 items that are CONCRETE bugs, broken features, stale "
|
||||
"state, or clear improvement opportunities in your own codebase.\n\n"
|
||||
"Rules:\n"
|
||||
"- Only include things that could become a real code fix or feature\n"
|
||||
"- Skip vague reflections, philosophical musings, or repeated themes\n"
|
||||
"- Each item needs a specific, descriptive title and body\n"
|
||||
"- Category must be one of: bug, feature, suggestion, maintenance\n\n"
|
||||
"Return ONLY a JSON array of objects with keys: "
|
||||
'"title", "body", "category"\n'
|
||||
"Return [] if nothing is actionable.\n\n"
|
||||
f"Recent thoughts:\n{thought_text}\n\nJSON array:"
|
||||
)
|
||||
|
||||
raw = self._call_agent(classify_prompt)
|
||||
if not raw or not raw.strip():
|
||||
return
|
||||
|
||||
import json
|
||||
|
||||
# Strip markdown code fences if present
|
||||
cleaned = raw.strip()
|
||||
if cleaned.startswith("```"):
|
||||
cleaned = cleaned.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
|
||||
|
||||
items = json.loads(cleaned)
|
||||
if not isinstance(items, list) or not items:
|
||||
return
|
||||
|
||||
from timmy.tools_gitea import create_gitea_issue
|
||||
|
||||
for item in items[:2]: # Safety cap
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
title = item.get("title", "").strip()
|
||||
body = item.get("body", "").strip()
|
||||
category = item.get("category", "suggestion").strip()
|
||||
if not title or len(title) < 10:
|
||||
continue
|
||||
|
||||
label = category if category in ("bug", "feature") else ""
|
||||
result = create_gitea_issue(title=title, body=body, labels=label)
|
||||
logger.info("Thought→Issue: %s → %s", title[:60], result[:80])
|
||||
|
||||
except Exception as exc:
|
||||
logger.debug("Thought issue filing skipped: %s", exc)
|
||||
|
||||
def _gather_system_snapshot(self) -> str:
|
||||
"""Gather lightweight real system state for grounding thoughts in reality.
|
||||
|
||||
|
||||
@@ -43,6 +43,8 @@ SAFE_TOOLS = frozenset(
|
||||
"check_ollama_health",
|
||||
"get_memory_status",
|
||||
"list_swarm_agents",
|
||||
"create_gitea_issue",
|
||||
"list_gitea_issues",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@@ -566,6 +566,16 @@ def create_full_toolkit(base_dir: str | Path | None = None):
|
||||
except Exception:
|
||||
logger.debug("Delegation tools not available")
|
||||
|
||||
# Gitea issue management — sovereign self-improvement channel
|
||||
try:
|
||||
from timmy.tools_gitea import create_gitea_issue, list_gitea_issues
|
||||
|
||||
toolkit.register(create_gitea_issue, name="create_gitea_issue")
|
||||
toolkit.register(list_gitea_issues, name="list_gitea_issues")
|
||||
logger.info("Gitea issue tools registered")
|
||||
except Exception:
|
||||
logger.debug("Gitea tools not available")
|
||||
|
||||
return toolkit
|
||||
|
||||
|
||||
|
||||
209
src/timmy/tools_gitea.py
Normal file
209
src/timmy/tools_gitea.py
Normal file
@@ -0,0 +1,209 @@
|
||||
"""Gitea tool functions — Timmy's self-improvement channel.
|
||||
|
||||
Provides sync tool wrappers around the async GiteaHand for use as
|
||||
Agno-registered agent tools. When Timmy notices a bug, stale state,
|
||||
or improvement opportunity, he can file a Gitea issue directly.
|
||||
|
||||
Usage::
|
||||
|
||||
from timmy.tools_gitea import create_gitea_issue, list_gitea_issues
|
||||
|
||||
# In agent conversation or thinking post-hook:
|
||||
result = create_gitea_issue(
|
||||
title="memory_forget tool returns error for valid queries",
|
||||
body="The memory_forget operation fails silently...",
|
||||
labels="bug",
|
||||
)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sqlite3
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _run_async(coro):
|
||||
"""Run an async coroutine from sync context (tool functions must be sync).
|
||||
|
||||
When no event loop is running, uses ``asyncio.run()``.
|
||||
When called from within an existing loop (e.g. FastAPI), spawns a
|
||||
new thread to avoid "cannot call asyncio.run from running loop".
|
||||
"""
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
# No running loop — safe to use asyncio.run()
|
||||
return asyncio.run(coro)
|
||||
|
||||
# Already in an async context — run in a new thread
|
||||
import concurrent.futures
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
||||
return pool.submit(asyncio.run, coro).result(timeout=60)
|
||||
|
||||
|
||||
def _bridge_to_work_order(title: str, body: str, category: str) -> None:
|
||||
"""Also create a local work order so the dashboard tracks it."""
|
||||
try:
|
||||
db_path = Path(settings.repo_root) / "data" / "work_orders.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS work_orders (
|
||||
id TEXT PRIMARY KEY,
|
||||
title TEXT NOT NULL,
|
||||
description TEXT DEFAULT '',
|
||||
priority TEXT DEFAULT 'medium',
|
||||
category TEXT DEFAULT 'suggestion',
|
||||
submitter TEXT DEFAULT 'dashboard',
|
||||
related_files TEXT DEFAULT '',
|
||||
status TEXT DEFAULT 'submitted',
|
||||
result TEXT DEFAULT '',
|
||||
rejection_reason TEXT DEFAULT '',
|
||||
created_at TEXT DEFAULT (datetime('now')),
|
||||
completed_at TEXT
|
||||
)"""
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO work_orders (id, title, description, category, submitter, created_at) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?)",
|
||||
(
|
||||
str(uuid.uuid4()),
|
||||
title,
|
||||
body,
|
||||
category,
|
||||
"timmy-thinking",
|
||||
datetime.utcnow().isoformat(),
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
except Exception as exc:
|
||||
logger.debug("Work order bridge failed: %s", exc)
|
||||
|
||||
|
||||
def create_gitea_issue(title: str, body: str = "", labels: str = "") -> str:
|
||||
"""Create an issue in the project's Gitea repository.
|
||||
|
||||
Use this when you notice a bug, broken feature, stale state, or
|
||||
improvement opportunity in your own codebase. Issues are tracked
|
||||
in Gitea and also bridged to the local work order queue.
|
||||
|
||||
Args:
|
||||
title: Short, descriptive issue title.
|
||||
body: Detailed description in markdown (symptoms, expected
|
||||
behaviour, relevant files).
|
||||
labels: Comma-separated label names (e.g. "bug,thinking-engine").
|
||||
|
||||
Returns:
|
||||
Confirmation with issue URL, or explanation if skipped/failed.
|
||||
"""
|
||||
from infrastructure.hands.gitea import gitea_hand
|
||||
|
||||
if not gitea_hand.available:
|
||||
return (
|
||||
"Gitea integration is not configured. "
|
||||
"Set GITEA_TOKEN or place a token in ~/.config/gitea/token."
|
||||
)
|
||||
|
||||
label_list = [tag.strip() for tag in labels.split(",") if tag.strip()] if labels else []
|
||||
|
||||
async def _create():
|
||||
# Dedup check — don't file if a similar issue is already open
|
||||
duplicate = await gitea_hand.find_duplicate(title)
|
||||
if duplicate:
|
||||
number = duplicate.get("number", "?")
|
||||
url = duplicate.get("html_url", "")
|
||||
return f"Skipped — similar issue already open: #{number} ({url})"
|
||||
|
||||
# Append auto-filing signature
|
||||
full_body = body
|
||||
if full_body:
|
||||
full_body += "\n\n"
|
||||
full_body += "---\n🤖 *Auto-filed by Timmy's thinking engine*"
|
||||
|
||||
result = await gitea_hand.create_issue(
|
||||
title=title,
|
||||
body=full_body,
|
||||
labels=label_list or None,
|
||||
)
|
||||
|
||||
if not result.success:
|
||||
return f"Failed to create issue: {result.error}"
|
||||
|
||||
issue_number = result.data.get("number", "?")
|
||||
issue_url = result.data.get("html_url", "")
|
||||
|
||||
# Bridge to local work order system
|
||||
category = "bug" if "bug" in label_list else "suggestion"
|
||||
_bridge_to_work_order(title, body, category)
|
||||
|
||||
# Emit event if bus is available
|
||||
try:
|
||||
from infrastructure.events.bus import emit
|
||||
|
||||
asyncio.ensure_future(
|
||||
emit(
|
||||
"gitea.issue.created",
|
||||
source="timmy-tools",
|
||||
data={
|
||||
"issue_number": issue_number,
|
||||
"title": title,
|
||||
"url": issue_url,
|
||||
},
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
logger.info("Created Gitea issue #%s: %s", issue_number, title)
|
||||
return f"Created issue #{issue_number}: {title}\n{issue_url}"
|
||||
|
||||
return _run_async(_create())
|
||||
|
||||
|
||||
def list_gitea_issues(state: str = "open") -> str:
|
||||
"""List issues in the project's Gitea repository.
|
||||
|
||||
Use this to check what issues are already filed before creating
|
||||
new ones, or to review the current backlog.
|
||||
|
||||
Args:
|
||||
state: Filter by state — "open" (default), "closed", or "all".
|
||||
|
||||
Returns:
|
||||
Formatted list of issues with number, title, and labels.
|
||||
"""
|
||||
from infrastructure.hands.gitea import gitea_hand
|
||||
|
||||
if not gitea_hand.available:
|
||||
return "Gitea integration is not configured."
|
||||
|
||||
async def _list():
|
||||
result = await gitea_hand.list_issues(state=state, limit=25)
|
||||
if not result.success:
|
||||
return f"Failed to list issues: {result.error}"
|
||||
|
||||
issues = result.data
|
||||
if not isinstance(issues, list) or not issues:
|
||||
return f"No {state} issues found."
|
||||
|
||||
lines = [f"## {state.capitalize()} Issues ({len(issues)})\n"]
|
||||
for issue in issues:
|
||||
number = issue.get("number", "?")
|
||||
title = issue.get("title", "Untitled")
|
||||
labels = ", ".join(label.get("name", "") for label in issue.get("labels", []))
|
||||
label_str = f" [{labels}]" if labels else ""
|
||||
lines.append(f"- **#{number}** {title}{label_str}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
return _run_async(_list())
|
||||
352
tests/test_hands_gitea.py
Normal file
352
tests/test_hands_gitea.py
Normal file
@@ -0,0 +1,352 @@
|
||||
"""Tests for the Gitea Hand.
|
||||
|
||||
Covers:
|
||||
- GiteaResult dataclass defaults
|
||||
- Token resolution (settings vs filesystem fallback)
|
||||
- Availability checks
|
||||
- Title similarity dedup
|
||||
- HTTP request handling (mocked)
|
||||
- Info summary
|
||||
- Graceful degradation on failure
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GiteaResult dataclass
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_gitea_result_defaults():
|
||||
"""GiteaResult should have sensible defaults."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
r = GiteaResult(operation="GET /issues", success=True)
|
||||
assert r.data == {}
|
||||
assert r.error == ""
|
||||
assert r.latency_ms == 0.0
|
||||
|
||||
|
||||
def test_gitea_result_with_data():
|
||||
"""GiteaResult should carry data."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
r = GiteaResult(
|
||||
operation="POST /issues",
|
||||
success=True,
|
||||
data={"number": 42, "html_url": "http://localhost:3000/issues/42"},
|
||||
latency_ms=15.3,
|
||||
)
|
||||
assert r.data["number"] == 42
|
||||
assert r.success is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Title similarity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_title_similar_matches():
|
||||
"""Similar titles should be detected as duplicates."""
|
||||
from infrastructure.hands.gitea import _title_similar
|
||||
|
||||
assert _title_similar("memory_forget tool fails", "Memory_forget tool fails") is True
|
||||
assert (
|
||||
_title_similar(
|
||||
"MEMORY.md not updating after thoughts",
|
||||
"MEMORY.md hasn't updated since March 8",
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
|
||||
def test_title_similar_rejects_different():
|
||||
"""Different titles should not match."""
|
||||
from infrastructure.hands.gitea import _title_similar
|
||||
|
||||
assert _title_similar("fix login page CSS", "add dark mode toggle") is False
|
||||
assert _title_similar("memory bug", "completely different topic") is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Token resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_resolve_token_from_settings():
|
||||
"""Token from settings should be preferred."""
|
||||
from infrastructure.hands.gitea import _resolve_token
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_token = "from-settings"
|
||||
assert _resolve_token() == "from-settings"
|
||||
|
||||
|
||||
def test_resolve_token_from_file(tmp_path):
|
||||
"""Token should fall back to filesystem."""
|
||||
from infrastructure.hands.gitea import _resolve_token
|
||||
|
||||
token_file = tmp_path / "token"
|
||||
token_file.write_text("from-file\n")
|
||||
|
||||
with (
|
||||
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
||||
patch("infrastructure.hands.gitea._TOKEN_FILE", token_file),
|
||||
):
|
||||
mock_settings.gitea_token = ""
|
||||
assert _resolve_token() == "from-file"
|
||||
|
||||
|
||||
def test_resolve_token_missing(tmp_path):
|
||||
"""Empty string when no token available."""
|
||||
|
||||
from infrastructure.hands.gitea import _resolve_token
|
||||
|
||||
with (
|
||||
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
||||
patch("infrastructure.hands.gitea._TOKEN_FILE", tmp_path / "nonexistent"),
|
||||
):
|
||||
mock_settings.gitea_token = ""
|
||||
assert _resolve_token() == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Availability
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_available_when_configured():
|
||||
"""Hand should be available when token and repo are set."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
assert hand.available is True
|
||||
|
||||
|
||||
def test_not_available_without_token():
|
||||
"""Hand should not be available without token."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
with (
|
||||
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
||||
patch("infrastructure.hands.gitea._resolve_token", return_value=""),
|
||||
):
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = ""
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="")
|
||||
assert hand.available is False
|
||||
|
||||
|
||||
def test_not_available_when_disabled():
|
||||
"""Hand should not be available when disabled."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = False
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
assert hand.available is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP request handling
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_returns_error_when_unavailable():
|
||||
"""_request should return error when not configured."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = False
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = ""
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="")
|
||||
result = await hand._request("GET", "/test")
|
||||
assert result.success is False
|
||||
assert "not configured" in result.error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_success():
|
||||
"""create_issue should POST and return issue data."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 201
|
||||
mock_response.text = '{"number": 1, "html_url": "http://localhost:3000/issues/1"}'
|
||||
mock_response.json.return_value = {
|
||||
"number": 1,
|
||||
"html_url": "http://localhost:3000/issues/1",
|
||||
}
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.request = AsyncMock(return_value=mock_response)
|
||||
mock_client.is_closed = False
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
hand._client = mock_client
|
||||
|
||||
result = await hand.create_issue("Test issue", "Body text")
|
||||
assert result.success is True
|
||||
assert result.data["number"] == 1
|
||||
|
||||
# Verify the API call
|
||||
mock_client.request.assert_called_once()
|
||||
call_args = mock_client.request.call_args
|
||||
assert call_args[0] == ("POST", "/api/v1/repos/owner/repo/issues")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_handles_http_error():
|
||||
"""create_issue should handle HTTP errors gracefully."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 500
|
||||
mock_response.text = "Internal Server Error"
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.request = AsyncMock(return_value=mock_response)
|
||||
mock_client.is_closed = False
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
hand._client = mock_client
|
||||
|
||||
result = await hand.create_issue("Test issue")
|
||||
assert result.success is False
|
||||
assert "500" in result.error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_handles_connection_error():
|
||||
"""create_issue should handle connection errors gracefully."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.request = AsyncMock(side_effect=ConnectionError("refused"))
|
||||
mock_client.is_closed = False
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
hand._client = mock_client
|
||||
|
||||
result = await hand.create_issue("Test issue")
|
||||
assert result.success is False
|
||||
assert "refused" in result.error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_duplicate_detects_match():
|
||||
"""find_duplicate should detect similar open issues."""
|
||||
from infrastructure.hands.gitea import GiteaHand, GiteaResult
|
||||
|
||||
existing_issues = [
|
||||
{"number": 5, "title": "MEMORY.md not updating", "html_url": "http://example.com/5"},
|
||||
{"number": 6, "title": "Add dark mode", "html_url": "http://example.com/6"},
|
||||
]
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
|
||||
# Mock list_issues
|
||||
hand.list_issues = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="GET",
|
||||
success=True,
|
||||
data=existing_issues,
|
||||
)
|
||||
)
|
||||
|
||||
dup = await hand.find_duplicate("MEMORY.md hasn't updated")
|
||||
assert dup is not None
|
||||
assert dup["number"] == 5
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_find_duplicate_no_match():
|
||||
"""find_duplicate should return None when no similar issue exists."""
|
||||
from infrastructure.hands.gitea import GiteaHand, GiteaResult
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
|
||||
hand.list_issues = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="GET",
|
||||
success=True,
|
||||
data=[
|
||||
{"number": 1, "title": "Completely unrelated issue"},
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
dup = await hand.find_duplicate("memory_forget tool throws error")
|
||||
assert dup is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Info summary
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_info_returns_summary():
|
||||
"""info() should return a dict with status."""
|
||||
from infrastructure.hands.gitea import GiteaHand
|
||||
|
||||
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
mock_settings.gitea_token = "test-token"
|
||||
mock_settings.gitea_repo = "owner/repo"
|
||||
mock_settings.gitea_timeout = 30
|
||||
hand = GiteaHand(token="test-token")
|
||||
info = hand.info()
|
||||
assert "base_url" in info
|
||||
assert "repo" in info
|
||||
assert "available" in info
|
||||
assert info["available"] is True
|
||||
215
tests/timmy/test_tools_gitea.py
Normal file
215
tests/timmy/test_tools_gitea.py
Normal file
@@ -0,0 +1,215 @@
|
||||
"""Tests for Gitea tool functions.
|
||||
|
||||
Covers:
|
||||
- create_gitea_issue tool (success, dedup skip, unavailable)
|
||||
- list_gitea_issues tool (success, empty, unavailable)
|
||||
- Work order bridge
|
||||
- Tool safety classification
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool safety classification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_gitea_tools_are_safe():
|
||||
"""Gitea tools should be classified as safe (no confirmation needed)."""
|
||||
from timmy.tool_safety import requires_confirmation
|
||||
|
||||
assert requires_confirmation("create_gitea_issue") is False
|
||||
assert requires_confirmation("list_gitea_issues") is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_gitea_issue
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# All patches target infrastructure.hands.gitea.gitea_hand because
|
||||
# tools_gitea.py uses deferred imports inside function bodies.
|
||||
|
||||
|
||||
def test_create_issue_unavailable():
|
||||
"""Should return message when Gitea is not configured."""
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = False
|
||||
|
||||
with patch("infrastructure.hands.gitea.gitea_hand", mock_hand):
|
||||
from timmy.tools_gitea import create_gitea_issue
|
||||
|
||||
result = create_gitea_issue("Test issue", "Body")
|
||||
assert "not configured" in result
|
||||
|
||||
|
||||
def test_create_issue_success():
|
||||
"""Should create issue and return confirmation."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = True
|
||||
mock_hand.find_duplicate = AsyncMock(return_value=None)
|
||||
mock_hand.create_issue = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="POST",
|
||||
success=True,
|
||||
data={"number": 42, "html_url": "http://localhost:3000/issues/42"},
|
||||
)
|
||||
)
|
||||
|
||||
with (
|
||||
patch("infrastructure.hands.gitea.gitea_hand", mock_hand),
|
||||
patch("timmy.tools_gitea._bridge_to_work_order"),
|
||||
):
|
||||
from timmy.tools_gitea import create_gitea_issue
|
||||
|
||||
result = create_gitea_issue("Test bug", "Bug description", "bug")
|
||||
assert "#42" in result
|
||||
assert "Test bug" in result
|
||||
|
||||
|
||||
def test_create_issue_dedup_skip():
|
||||
"""Should skip when similar issue already exists."""
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = True
|
||||
mock_hand.find_duplicate = AsyncMock(
|
||||
return_value={"number": 10, "html_url": "http://localhost:3000/issues/10"}
|
||||
)
|
||||
|
||||
with patch("infrastructure.hands.gitea.gitea_hand", mock_hand):
|
||||
from timmy.tools_gitea import create_gitea_issue
|
||||
|
||||
result = create_gitea_issue("Existing issue")
|
||||
assert "Skipped" in result
|
||||
assert "#10" in result
|
||||
|
||||
|
||||
def test_create_issue_api_failure():
|
||||
"""Should return error message on API failure."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = True
|
||||
mock_hand.find_duplicate = AsyncMock(return_value=None)
|
||||
mock_hand.create_issue = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="POST",
|
||||
success=False,
|
||||
error="HTTP 500: Internal Server Error",
|
||||
)
|
||||
)
|
||||
|
||||
with (
|
||||
patch("infrastructure.hands.gitea.gitea_hand", mock_hand),
|
||||
patch("timmy.tools_gitea._bridge_to_work_order"),
|
||||
):
|
||||
from timmy.tools_gitea import create_gitea_issue
|
||||
|
||||
result = create_gitea_issue("Test issue")
|
||||
assert "Failed" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# list_gitea_issues
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_list_issues_unavailable():
|
||||
"""Should return message when Gitea is not configured."""
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = False
|
||||
|
||||
with patch("infrastructure.hands.gitea.gitea_hand", mock_hand):
|
||||
from timmy.tools_gitea import list_gitea_issues
|
||||
|
||||
result = list_gitea_issues()
|
||||
assert "not configured" in result
|
||||
|
||||
|
||||
def test_list_issues_success():
|
||||
"""Should return formatted issue list."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = True
|
||||
mock_hand.list_issues = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="GET",
|
||||
success=True,
|
||||
data=[
|
||||
{"number": 1, "title": "Bug fix", "labels": [{"name": "bug"}]},
|
||||
{"number": 2, "title": "Feature request", "labels": []},
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
with patch("infrastructure.hands.gitea.gitea_hand", mock_hand):
|
||||
from timmy.tools_gitea import list_gitea_issues
|
||||
|
||||
result = list_gitea_issues("open")
|
||||
assert "#1" in result
|
||||
assert "Bug fix" in result
|
||||
assert "[bug]" in result
|
||||
assert "#2" in result
|
||||
|
||||
|
||||
def test_list_issues_empty():
|
||||
"""Should return empty message when no issues."""
|
||||
from infrastructure.hands.gitea import GiteaResult
|
||||
|
||||
mock_hand = MagicMock()
|
||||
mock_hand.available = True
|
||||
mock_hand.list_issues = AsyncMock(
|
||||
return_value=GiteaResult(
|
||||
operation="GET",
|
||||
success=True,
|
||||
data=[],
|
||||
)
|
||||
)
|
||||
|
||||
with patch("infrastructure.hands.gitea.gitea_hand", mock_hand):
|
||||
from timmy.tools_gitea import list_gitea_issues
|
||||
|
||||
result = list_gitea_issues()
|
||||
assert "No open issues" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Work order bridge
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_bridge_to_work_order(tmp_path):
|
||||
"""Should create a work order in the local database."""
|
||||
from timmy.tools_gitea import _bridge_to_work_order
|
||||
|
||||
# Point to a "data" subdir inside tmp_path so the code creates it
|
||||
data_dir = tmp_path / "data"
|
||||
data_dir.mkdir()
|
||||
db_path = data_dir / "work_orders.db"
|
||||
|
||||
with patch("timmy.tools_gitea.settings") as mock_settings:
|
||||
mock_settings.repo_root = str(tmp_path)
|
||||
_bridge_to_work_order("Test WO", "Description", "bug")
|
||||
|
||||
import sqlite3
|
||||
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.row_factory = sqlite3.Row
|
||||
rows = conn.execute("SELECT * FROM work_orders").fetchall()
|
||||
conn.close()
|
||||
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["title"] == "Test WO"
|
||||
assert rows[0]["submitter"] == "timmy-thinking"
|
||||
assert rows[0]["category"] == "bug"
|
||||
|
||||
|
||||
def test_bridge_to_work_order_graceful_failure():
|
||||
"""Should not raise when bridge fails."""
|
||||
from timmy.tools_gitea import _bridge_to_work_order
|
||||
|
||||
with patch("timmy.tools_gitea.settings") as mock_settings:
|
||||
mock_settings.repo_root = "/nonexistent/path/that/cannot/exist"
|
||||
# Should not raise
|
||||
_bridge_to_work_order("Test", "Desc", "bug")
|
||||
Reference in New Issue
Block a user