pelmen/kin
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: KIN-048 Post-pipeline hook: автокоммит после успешного завершения задачи. git add -A && git commit -m 'kin: TASK_ID TITLE'. Срабатывает автоматически как rebuild-frontend.

Browse source
This commit is contained in:
Gros Frumos 2026-03-16 06:59:46 +02:00
parent 8a6f280cbd
commit ae21e48b65
13 changed files with 1554 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

41
agents/prompts/learner.md Normal file
View file

@ -0,0 +1,41 @@
You are a learning extractor for the Kin multi-agent orchestrator.
Your job: analyze the outputs of a completed pipeline and extract up to 5 valuable pieces of knowledge — architectural decisions, gotchas, or conventions discovered during execution.
## Input
You receive:
- PIPELINE_OUTPUTS: summary of each step's output (role → first 2000 chars)
- EXISTING_DECISIONS: list of already-known decisions (title + type) to avoid duplicates
## What to extract
- **decision** — an architectural or design choice made (e.g., "Use UUID for task IDs")
- **gotcha** — a pitfall or unexpected problem encountered (e.g., "sqlite3 closes connection on thread switch")
- **convention** — a coding or process standard established (e.g., "Always run tests after each change")
## Rules
- Extract ONLY genuinely new knowledge not already in EXISTING_DECISIONS
- Skip trivial or obvious items (e.g., "write clean code")
- Skip task-specific results that won't generalize (e.g., "fixed bug in useSearch.ts line 42")
- Each decision must be actionable and reusable across future tasks
- Extract at most 5 decisions total; fewer is better than low-quality ones
- If nothing valuable found, return empty list
## Output format
Return ONLY valid JSON (no markdown, no explanation):
```json
{
"decisions": [
{
"type": "decision",
"title": "Short memorable title",
"description": "Clear explanation of what was decided and why",
"tags": ["optional", "tags"]
}
]
}
```

11
agents/prompts/pm.md
View file

@ -30,6 +30,16 @@ You receive:
- Don't assign specialists who aren't needed.
- If a task is blocked or unclear, say so — don't guess.
## Completion mode selection
Set `completion_mode` based on the following rules (in priority order):
1. If `project.execution_mode` is set — use it as the default.
2. Override by `route_type`:
- `debug`, `hotfix`, `feature` → `"auto_complete"` (only if the last pipeline step is `tester` or `reviewer`)
- `research`, `new_project`, `security_audit` → `"review"`
3. Fallback: `"review"`
## Output format
Return ONLY valid JSON (no markdown, no explanation):
@ -37,6 +47,7 @@ Return ONLY valid JSON (no markdown, no explanation):
```json
{
"analysis": "Brief analysis of what needs to be done",
"completion_mode": "auto_complete",
"pipeline": [
{
"role": "debugger",

252
agents/runner.py
View file

@ -4,7 +4,9 @@ Each agent = separate process with isolated context.
"""
import json
import logging
import os
import shutil
import sqlite3
import subprocess
import time
@ -13,6 +15,50 @@ from typing import Any
import re
_logger = logging.getLogger("kin.runner")
# Extra PATH entries to inject when searching for claude CLI.
# launchctl daemons start with a stripped PATH that may omit these.
_EXTRA_PATH_DIRS = [
"/opt/homebrew/bin",
"/opt/homebrew/sbin",
"/usr/local/bin",
"/usr/local/sbin",
]
def _build_claude_env() -> dict:
"""Return an env dict with an extended PATH that includes common CLI tool locations.
Merges _EXTRA_PATH_DIRS with the current process PATH, deduplicating entries.
Also resolves ~/.nvm/versions/node/*/bin globs that launchctl may not expand.
"""
env = os.environ.copy()
existing = env.get("PATH", "").split(":")
extra = list(_EXTRA_PATH_DIRS)
# Expand nvm node bin dirs dynamically
nvm_root = Path.home() / ".nvm" / "versions" / "node"
if nvm_root.is_dir():
for node_ver in sorted(nvm_root.iterdir(), reverse=True):
bin_dir = node_ver / "bin"
if bin_dir.is_dir():
extra.append(str(bin_dir))
seen = set(existing)
new_dirs = [d for d in extra if d and d not in seen]
env["PATH"] = ":".join(new_dirs + existing)
return env
def _resolve_claude_cmd() -> str:
"""Return the full path to the claude CLI, or 'claude' as fallback."""
extended_env = _build_claude_env()
found = shutil.which("claude", path=extended_env["PATH"])
return found or "claude"
from core import models
from core.context_builder import build_context, format_prompt
from core.hooks import run_hooks
@ -116,10 +162,12 @@ def _run_claude(
working_dir: str | None = None,
allow_write: bool = False,
noninteractive: bool = False,
timeout: int | None = None,
) -> dict:
"""Execute claude CLI as subprocess. Returns dict with output, returncode, etc."""
claude_cmd = _resolve_claude_cmd()
cmd = [
"claude",
claude_cmd,
"-p", prompt,
"--output-format", "json",
"--model", model,
@ -128,7 +176,9 @@ def _run_claude(
cmd.append("--dangerously-skip-permissions")
is_noninteractive = noninteractive or os.environ.get("KIN_NONINTERACTIVE") == "1"
timeout = 300 if is_noninteractive else 600
if timeout is None:
timeout = int(os.environ.get("KIN_AGENT_TIMEOUT") or 600)
env = _build_claude_env()
try:
proc = subprocess.run(
@ -137,6 +187,7 @@ def _run_claude(
text=True,
timeout=timeout,
cwd=working_dir,
env=env,
stdin=subprocess.DEVNULL if is_noninteractive else None,
)
except FileNotFoundError:
@ -377,6 +428,179 @@ def _is_permission_error(result: dict) -> bool:
return any(re.search(p, text) for p in PERMISSION_PATTERNS)
# ---------------------------------------------------------------------------
# Autocommit: git add -A && git commit after successful pipeline
# ---------------------------------------------------------------------------
def _run_autocommit(
conn: sqlite3.Connection,
task_id: str,
project_id: str,
) -> None:
"""Auto-commit changes after successful pipeline completion.
Runs: git add -A && git commit -m 'kin: {task_id} {title}'.
Silently skips if nothing to commit (exit code 1) or project path not found.
Never raises — autocommit errors must never block the pipeline.
Uses stderr=subprocess.DEVNULL per decision #30.
"""
task = models.get_task(conn, task_id)
project = models.get_project(conn, project_id)
if not task or not project:
return
if not project.get("autocommit_enabled"):
return
project_path = Path(project["path"]).expanduser()
if not project_path.is_dir():
return
working_dir = str(project_path)
env = _build_claude_env()
git_cmd = shutil.which("git", path=env["PATH"]) or "git"
title = (task.get("title") or "").replace('"', "'").replace("\n", " ").replace("\r", "")
commit_msg = f"kin: {task_id} {title}"
try:
subprocess.run(
[git_cmd, "add", "-A"],
cwd=working_dir,
env=env,
stderr=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
)
result = subprocess.run(
[git_cmd, "commit", "-m", commit_msg],
cwd=working_dir,
env=env,
stderr=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
)
if result.returncode == 0:
_logger.info("Autocommit: %s", commit_msg)
else:
_logger.debug("Autocommit: nothing to commit for %s", task_id)
except Exception as exc:
_logger.warning("Autocommit failed for %s: %s", task_id, exc)
# ---------------------------------------------------------------------------
# Auto-learning: extract decisions from pipeline results
# ---------------------------------------------------------------------------
VALID_DECISION_TYPES = {"decision", "gotcha", "convention"}
def _run_learning_extraction(
conn: sqlite3.Connection,
task_id: str,
project_id: str,
step_results: list[dict],
) -> dict:
"""Extract and save decisions from completed pipeline results.
Calls the learner agent with step outputs + existing decisions,
parses the JSON response, and saves new decisions via add_decision_if_new.
Returns a summary dict with added/skipped counts.
"""
learner_prompt_path = PROMPTS_DIR / "learner.md"
if not learner_prompt_path.exists():
return {"added": 0, "skipped": 0, "error": "learner.md not found"}
template = learner_prompt_path.read_text()
# Summarize step outputs (first 2000 chars each)
step_summaries = {}
for r in step_results:
role = r.get("role", "unknown")
output = r.get("raw_output") or r.get("output") or ""
if isinstance(output, (dict, list)):
output = json.dumps(output, ensure_ascii=False)
step_summaries[role] = output[:2000]
# Fetch existing decisions for dedup hint
existing = models.get_decisions(conn, project_id)
existing_hints = [
{"title": d["title"], "type": d["type"]}
for d in existing
]
prompt_parts = [
template,
"",
"## PIPELINE_OUTPUTS",
json.dumps(step_summaries, ensure_ascii=False, indent=2),
"",
"## EXISTING_DECISIONS",
json.dumps(existing_hints, ensure_ascii=False, indent=2),
]
prompt = "\n".join(prompt_parts)
learner_timeout = int(os.environ.get("KIN_LEARNER_TIMEOUT") or 120)
start = time.monotonic()
result = _run_claude(prompt, model="sonnet", noninteractive=True, timeout=learner_timeout)
duration = int(time.monotonic() - start)
raw_output = result.get("output", "")
if not isinstance(raw_output, str):
raw_output = json.dumps(raw_output, ensure_ascii=False)
success = result["returncode"] == 0
# Log to agent_logs
models.log_agent_run(
conn,
project_id=project_id,
task_id=task_id,
agent_role="learner",
action="learn",
input_summary=f"project={project_id}, task={task_id}, steps={len(step_results)}",
output_summary=raw_output or None,
tokens_used=result.get("tokens_used"),
model="sonnet",
cost_usd=result.get("cost_usd"),
success=success,
error_message=result.get("error") if not success else None,
duration_seconds=duration,
)
parsed = _try_parse_json(raw_output)
if not isinstance(parsed, dict):
return {"added": 0, "skipped": 0, "error": "non-JSON learner output"}
decisions = parsed.get("decisions", [])
if not isinstance(decisions, list):
return {"added": 0, "skipped": 0, "error": "invalid decisions format"}
added = 0
skipped = 0
for item in decisions[:5]:
if not isinstance(item, dict):
continue
d_type = item.get("type", "decision")
if d_type not in VALID_DECISION_TYPES:
d_type = "decision"
d_title = (item.get("title") or "").strip()
d_desc = (item.get("description") or "").strip()
if not d_title or not d_desc:
continue
saved = models.add_decision_if_new(
conn,
project_id=project_id,
type=d_type,
title=d_title,
description=d_desc,
tags=item.get("tags") or [],
task_id=task_id,
)
if saved:
added += 1
else:
skipped += 1
return {"added": added, "skipped": skipped}
# ---------------------------------------------------------------------------
# Pipeline executor
# ---------------------------------------------------------------------------
@ -485,7 +709,7 @@ def run_pipeline(
if not result["success"]:
# Auto mode: retry once with allow_write on permission error
if mode == "auto" and not allow_write and _is_permission_error(result):
if mode == "auto_complete" and not allow_write and _is_permission_error(result):
task_modules = models.get_modules(conn, project_id)
try:
run_hooks(conn, project_id, task_id,
@ -555,8 +779,11 @@ def run_pipeline(
task_modules = models.get_modules(conn, project_id)
if mode == "auto":
# Auto mode: skip review, approve immediately
last_role = steps[-1].get("role", "") if steps else ""
auto_eligible = last_role in {"tester", "reviewer"}
if mode == "auto_complete" and auto_eligible:
# Auto-complete mode: last step is tester/reviewer — skip review, approve immediately
models.update_task(conn, task_id, status="done")
try:
run_hooks(conn, project_id, task_id,
@ -586,7 +813,7 @@ def run_pipeline(
pass
else:
# Review mode: wait for manual approval
models.update_task(conn, task_id, status="review")
models.update_task(conn, task_id, status="review", execution_mode="review")
# Run post-pipeline hooks (failures don't affect pipeline status)
try:
@ -595,6 +822,19 @@ def run_pipeline(
except Exception:
pass # Hook errors must never block pipeline completion
# Auto-learning: extract decisions from pipeline results
if results:
try:
_run_learning_extraction(conn, task_id, project_id, results)
except Exception:
pass # Learning errors must never block pipeline completion
# Auto-commit changes after successful pipeline
try:
_run_autocommit(conn, task_id, project_id)
except Exception:
pass # Autocommit errors must never block pipeline completion
return {
"success": True,
"steps_completed": len(steps),

12
cli/main.py
View file

@ -586,6 +586,18 @@ def run_task(ctx, task_id, dry_run, allow_write):
pipeline_steps = output["pipeline"]
analysis = output.get("analysis", "")
# Save completion_mode from PM output to task (only if not already set by user)
task_current = models.get_task(conn, task_id)
if not task_current.get("execution_mode"):
pm_completion_mode = models.validate_completion_mode(
output.get("completion_mode", "review")
)
models.update_task(conn, task_id, execution_mode=pm_completion_mode)
import logging
logging.getLogger("kin").info(
"PM set completion_mode=%s for task %s", pm_completion_mode, task_id
)
click.echo(f"\nAnalysis: {analysis}")
click.echo(f"Pipeline ({len(pipeline_steps)} steps):")
for i, step in enumerate(pipeline_steps, 1):

1
core/context_builder.py
View file

@ -110,6 +110,7 @@ def _slim_project(project: dict) -> dict:
"path": project["path"],
"tech_stack": project.get("tech_stack"),
"language": project.get("language", "ru"),
"execution_mode": project.get("execution_mode"),
}

49
core/db.py
View file

@ -216,12 +216,61 @@ def _migrate(conn: sqlite3.Connection):
conn.execute("ALTER TABLE tasks ADD COLUMN blocked_reason TEXT")
conn.commit()
if "autocommit_enabled" not in proj_cols:
conn.execute("ALTER TABLE projects ADD COLUMN autocommit_enabled INTEGER DEFAULT 0")
conn.commit()
# Rename legacy 'auto' → 'auto_complete' (KIN-063)
conn.execute(
"UPDATE projects SET execution_mode = 'auto_complete' WHERE execution_mode = 'auto'"
)
conn.execute(
"UPDATE tasks SET execution_mode = 'auto_complete' WHERE execution_mode = 'auto'"
)
conn.commit()
def _seed_default_hooks(conn: sqlite3.Connection):
"""Seed default hooks for the kin project (idempotent).
Creates rebuild-frontend hook only when:
- project 'kin' exists in the projects table
- the hook doesn't already exist (no duplicate)
"""
kin_exists = conn.execute(
"SELECT 1 FROM projects WHERE id = 'kin'"
).fetchone()
if not kin_exists:
return
exists = conn.execute(
"SELECT 1 FROM hooks"
" WHERE project_id = 'kin'"
" AND name = 'rebuild-frontend'"
" AND event = 'pipeline_completed'"
).fetchone()
if not exists:
conn.execute(
"""INSERT INTO hooks (project_id, name, event, command, enabled)
VALUES ('kin', 'rebuild-frontend', 'pipeline_completed',
'cd /Users/grosfrumos/projects/kin/web/frontend && npm run build',
1)"""
)
conn.commit()
# Enable autocommit for kin project (opt-in, idempotent)
conn.execute(
"UPDATE projects SET autocommit_enabled=1 WHERE id='kin' AND autocommit_enabled=0"
)
conn.commit()
def init_db(db_path: Path = DB_PATH) -> sqlite3.Connection:
conn = get_connection(db_path)
conn.executescript(SCHEMA)
conn.commit()
_migrate(conn)
_seed_default_hooks(conn)
return conn

35
core/models.py
View file

@ -14,6 +14,15 @@ VALID_TASK_STATUSES = [
"blocked", "decomposed", "cancelled",
]
VALID_COMPLETION_MODES = {"auto_complete", "review"}
def validate_completion_mode(value: str) -> str:
"""Validate completion mode from LLM output. Falls back to 'review' if invalid."""
if value in VALID_COMPLETION_MODES:
return value
return "review"
def _row_to_dict(row: sqlite3.Row | None) -> dict | None:
"""Convert sqlite3.Row to dict with JSON fields decoded."""
@ -220,6 +229,32 @@ def add_decision(
return _row_to_dict(row)
def add_decision_if_new(
conn: sqlite3.Connection,
project_id: str,
type: str,
title: str,
description: str,
category: str | None = None,
tags: list | None = None,
task_id: str | None = None,
) -> dict | None:
"""Add a decision only if no existing one matches (project_id, type, normalized title).
Returns the new decision dict, or None if skipped as duplicate.
"""
existing = conn.execute(
"""SELECT id FROM decisions
WHERE project_id = ? AND type = ?
AND lower(trim(title)) = lower(trim(?))""",
(project_id, type, title),
).fetchone()
if existing:
return None
return add_decision(conn, project_id, type, title, description,
category=category, tags=tags, task_id=task_id)
def get_decisions(
conn: sqlite3.Connection,
project_id: str,

37
tests/test_api.py
View file

@ -342,6 +342,19 @@ def test_patch_task_empty_body_returns_400(client):
assert r.status_code == 400
def test_patch_task_execution_mode_auto_complete_accepted(client):
"""KIN-063: execution_mode='auto_complete' принимается (200)."""
r = client.patch("/api/tasks/P1-001", json={"execution_mode": "auto_complete"})
assert r.status_code == 200
assert r.json()["execution_mode"] == "auto_complete"
def test_patch_task_execution_mode_auto_rejected(client):
"""KIN-063: старое значение 'auto' должно отклоняться (400) — Decision #29."""
r = client.patch("/api/tasks/P1-001", json={"execution_mode": "auto"})
assert r.status_code == 400
# ---------------------------------------------------------------------------
# KIN-022 — blocked_reason: регрессионные тесты
# ---------------------------------------------------------------------------
@ -589,3 +602,27 @@ def test_run_kin_040_allow_write_true_ignored(client):
Эндпоинт не имеет body-параметра, поэтому FastAPI не валидирует тело."""
r = client.post("/api/tasks/P1-001/run", json={"allow_write": True})
assert r.status_code == 202
# ---------------------------------------------------------------------------
# KIN-058 — регрессионный тест: stderr=DEVNULL у Popen в web API
# ---------------------------------------------------------------------------
def test_run_sets_stderr_devnull(client):
"""Регрессионный тест KIN-058: stderr=DEVNULL всегда устанавливается в Popen,
чтобы stderr дочернего процесса не загрязнял логи uvicorn."""
import subprocess as _subprocess
from unittest.mock import patch, MagicMock
with patch("web.api.subprocess.Popen") as mock_popen:
mock_proc = MagicMock()
mock_proc.pid = 77
mock_popen.return_value = mock_proc
r = client.post("/api/tasks/P1-001/run")
assert r.status_code == 202
call_kwargs = mock_popen.call_args[1]
assert call_kwargs.get("stderr") == _subprocess.DEVNULL, (
"Регрессия KIN-058: stderr у Popen должен быть DEVNULL, "
"иначе вывод агента попадает в логи uvicorn"
)

77
tests/test_auto_mode.py
View file

@ -1,7 +1,8 @@
"""
Tests for KIN-012 auto mode features:
Tests for KIN-012/KIN-063 auto mode features:
- TestAutoApprove: pipeline auto-approves (status → done) без ручного review
(KIN-063: auto_complete только если последний шаг — tester или reviewer)
- TestAutoRerunOnPermissionDenied: runner делает retry при permission error,
останавливается после одного retry (лимит = 1)
- TestAutoFollowup: generate_followups вызывается сразу, без ожидания
@ -75,30 +76,30 @@ class TestAutoApprove:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_sets_status_done(self, mock_run, mock_hooks, mock_followup, conn):
"""Auto-режим: статус задачи становится 'done', а не 'review'."""
"""Auto-complete режим: статус становится 'done', если последний шаг — tester."""
mock_run.return_value = _mock_success()
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find bug"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find bug"}, {"role": "tester", "brief": "verify fix"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "done", "Auto-mode должен auto-approve: status=done"
assert task["status"] == "done", "Auto-complete должен auto-approve: status=done"
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_fires_task_auto_approved_hook(self, mock_run, mock_hooks, mock_followup, conn):
"""В auto-режиме срабатывает хук task_auto_approved."""
"""В auto_complete-режиме срабатывает хук task_auto_approved (если последний шаг — tester)."""
mock_run.return_value = _mock_success()
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find bug"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find bug"}, {"role": "tester", "brief": "verify"}]
run_pipeline(conn, "VDOL-001", steps)
events = _get_hook_events(mock_hooks)
@ -140,20 +141,20 @@ class TestAutoApprove:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_task_level_auto_overrides_project_review(self, mock_run, mock_hooks, mock_followup, conn):
"""Если у задачи execution_mode=auto, pipeline auto-approve, даже если проект в review."""
"""Если у задачи execution_mode=auto_complete, pipeline auto-approve, даже если проект в review."""
mock_run.return_value = _mock_success()
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
# Проект в review, но задача — auto
models.update_task(conn, "VDOL-001", execution_mode="auto")
# Проект в review, но задача — auto_complete
models.update_task(conn, "VDOL-001", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}]
steps = [{"role": "debugger", "brief": "find"}, {"role": "reviewer", "brief": "approve"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "done", "Task-level auto должен override project review"
assert task["status"] == "done", "Task-level auto_complete должен override project review"
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@ -164,11 +165,11 @@ class TestAutoApprove:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result.get("mode") == "auto"
assert result.get("mode") == "auto_complete"
# ---------------------------------------------------------------------------
@ -178,10 +179,12 @@ class TestAutoApprove:
class TestAutoRerunOnPermissionDenied:
"""Runner повторяет шаг при permission issues, останавливается по лимиту (1 retry)."""
@patch("agents.runner._run_autocommit")
@patch("agents.runner._run_learning_extraction")
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_retries_on_permission_error(self, mock_run, mock_hooks, mock_followup, conn):
def test_auto_mode_retries_on_permission_error(self, mock_run, mock_hooks, mock_followup, mock_learn, mock_autocommit, conn):
"""Auto-режим: при permission denied runner делает 1 retry с allow_write=True."""
mock_run.side_effect = [
_mock_permission_denied(), # 1-й вызов: permission error
@ -189,8 +192,9 @@ class TestAutoRerunOnPermissionDenied:
]
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
mock_learn.return_value = {"added": 0, "skipped": 0}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "fix file"}]
result = run_pipeline(conn, "VDOL-001", steps)
@ -209,7 +213,7 @@ class TestAutoRerunOnPermissionDenied:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "fix"}]
run_pipeline(conn, "VDOL-001", steps)
@ -229,7 +233,7 @@ class TestAutoRerunOnPermissionDenied:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "fix"}]
run_pipeline(conn, "VDOL-001", steps)
@ -248,7 +252,7 @@ class TestAutoRerunOnPermissionDenied:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "fix"}]
result = run_pipeline(conn, "VDOL-001", steps)
@ -257,10 +261,12 @@ class TestAutoRerunOnPermissionDenied:
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "blocked"
@patch("agents.runner._run_autocommit")
@patch("agents.runner._run_learning_extraction")
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_subsequent_steps_use_allow_write_after_retry(self, mock_run, mock_hooks, mock_followup, conn):
def test_subsequent_steps_use_allow_write_after_retry(self, mock_run, mock_hooks, mock_followup, mock_learn, mock_autocommit, conn):
"""После успешного retry все следующие шаги тоже используют allow_write."""
mock_run.side_effect = [
_mock_permission_denied(), # Шаг 1: permission error
@ -269,8 +275,9 @@ class TestAutoRerunOnPermissionDenied:
]
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
mock_learn.return_value = {"added": 0, "skipped": 0}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [
{"role": "debugger", "brief": "fix"},
{"role": "tester", "brief": "test"},
@ -293,7 +300,7 @@ class TestAutoRerunOnPermissionDenied:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "fix"}]
result = run_pipeline(conn, "VDOL-001", steps)
@ -330,13 +337,13 @@ class TestAutoFollowup:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_followup_triggered_immediately(self, mock_run, mock_hooks, mock_followup, conn):
"""В auto-режиме generate_followups вызывается сразу после pipeline."""
"""В auto_complete-режиме generate_followups вызывается сразу после pipeline (последний шаг — tester)."""
mock_run.return_value = _mock_success()
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -357,8 +364,8 @@ class TestAutoFollowup:
mock_followup.return_value = {"created": [], "pending_actions": pending}
mock_resolve.return_value = [{"resolved": "rerun", "result": {}}]
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
run_pipeline(conn, "VDOL-001", steps)
mock_resolve.assert_called_once_with(conn, "VDOL-001", pending)
@ -392,10 +399,10 @@ class TestAutoFollowup:
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
models.update_task(conn, "VDOL-001", brief={"source": "followup:VDOL-000"})
steps = [{"role": "debugger", "brief": "find"}]
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -412,8 +419,8 @@ class TestAutoFollowup:
mock_hooks.return_value = []
mock_followup.side_effect = Exception("followup PM crashed")
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True # Pipeline succeeded, followup failure absorbed
@ -431,8 +438,8 @@ class TestAutoFollowup:
mock_followup.return_value = {"created": [], "pending_actions": []}
mock_resolve.return_value = []
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
run_pipeline(conn, "VDOL-001", steps)
mock_resolve.assert_not_called()

112
tests/test_hooks.py
View file

@ -1,6 +1,8 @@
"""Tests for core/hooks.py — post-pipeline hook execution."""
import os
import subprocess
import tempfile
import pytest
from unittest.mock import patch, MagicMock
@ -539,10 +541,6 @@ class TestKIN052RebuildFrontendCommand:
Симулирует рестарт: создаём хук, закрываем соединение, открываем новое — хук на месте.
"""
import tempfile
import os
from core.db import init_db
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
@ -568,3 +566,109 @@ class TestKIN052RebuildFrontendCommand:
assert hooks[0]["trigger_module_path"] is None
finally:
os.unlink(db_path)
# ---------------------------------------------------------------------------
# KIN-053: _seed_default_hooks — автоматический хук при инициализации БД
# ---------------------------------------------------------------------------
class TestKIN053SeedDefaultHooks:
"""Тесты для _seed_default_hooks (KIN-053).
При init_db автоматически создаётся rebuild-frontend хук для проекта 'kin',
если этот проект уже существует в БД. Функция идемпотентна.
"""
def test_seed_skipped_when_no_kin_project(self):
"""_seed_default_hooks не создаёт хук, если проекта 'kin' нет."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
conn = init_db(db_path)
hooks = get_hooks(conn, "kin", enabled_only=False)
conn.close()
assert hooks == []
finally:
os.unlink(db_path)
def test_seed_creates_hook_when_kin_project_exists(self):
"""_seed_default_hooks создаёт rebuild-frontend хук при наличии проекта 'kin'.
Порядок: init_db → create_project('kin') → повторный init_db → хук есть.
"""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
conn1 = init_db(db_path)
models.create_project(conn1, "kin", "Kin", "/projects/kin")
conn1.close()
conn2 = init_db(db_path)
hooks = get_hooks(conn2, "kin", event="pipeline_completed", enabled_only=True)
conn2.close()
assert len(hooks) == 1
assert hooks[0]["name"] == "rebuild-frontend"
assert "npm run build" in hooks[0]["command"]
assert "web/frontend" in hooks[0]["command"]
finally:
os.unlink(db_path)
def test_seed_hook_has_correct_command(self):
"""Команда хука — точная строка с cd && npm run build."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
conn1 = init_db(db_path)
models.create_project(conn1, "kin", "Kin", "/projects/kin")
conn1.close()
conn2 = init_db(db_path)
hooks = get_hooks(conn2, "kin", event="pipeline_completed", enabled_only=False)
conn2.close()
assert hooks[0]["command"] == (
"cd /Users/grosfrumos/projects/kin/web/frontend && npm run build"
)
assert hooks[0]["trigger_module_path"] is None
finally:
os.unlink(db_path)
def test_seed_idempotent_no_duplicate(self):
"""Повторные вызовы init_db не дублируют хук."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
conn = init_db(db_path)
models.create_project(conn, "kin", "Kin", "/projects/kin")
conn.close()
for _ in range(3):
c = init_db(db_path)
c.close()
conn_final = init_db(db_path)
hooks = get_hooks(conn_final, "kin", event="pipeline_completed", enabled_only=False)
conn_final.close()
assert len(hooks) == 1, f"Ожидается 1 хук, получено {len(hooks)}"
finally:
os.unlink(db_path)
def test_seed_hook_does_not_affect_other_projects(self):
"""Seed не создаёт хуки для других проектов."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
try:
conn1 = init_db(db_path)
models.create_project(conn1, "kin", "Kin", "/projects/kin")
models.create_project(conn1, "other", "Other", "/projects/other")
conn1.close()
conn2 = init_db(db_path)
other_hooks = get_hooks(conn2, "other", enabled_only=False)
conn2.close()
assert other_hooks == []
finally:
os.unlink(db_path)

92
tests/test_models.py
View file

@ -53,6 +53,55 @@ def test_update_project_tech_stack_json(conn):
assert updated["tech_stack"] == ["python", "fastapi"]
# -- validate_completion_mode (KIN-063) --
def test_validate_completion_mode_valid_auto_complete():
"""validate_completion_mode принимает 'auto_complete'."""
assert models.validate_completion_mode("auto_complete") == "auto_complete"
def test_validate_completion_mode_valid_review():
"""validate_completion_mode принимает 'review'."""
assert models.validate_completion_mode("review") == "review"
def test_validate_completion_mode_invalid_fallback():
"""validate_completion_mode возвращает 'review' для невалидных значений (фоллбэк)."""
assert models.validate_completion_mode("auto") == "review"
assert models.validate_completion_mode("") == "review"
assert models.validate_completion_mode("unknown") == "review"
# -- get_effective_mode (KIN-063) --
def test_get_effective_mode_task_overrides_project(conn):
"""Task execution_mode имеет приоритет над project execution_mode."""
models.create_project(conn, "p1", "P1", "/p1", execution_mode="review")
models.create_task(conn, "P1-001", "p1", "Task", execution_mode="auto_complete")
mode = models.get_effective_mode(conn, "p1", "P1-001")
assert mode == "auto_complete"
def test_get_effective_mode_falls_back_to_project(conn):
"""Если задача без execution_mode — применяется project execution_mode."""
models.create_project(conn, "p1", "P1", "/p1", execution_mode="auto_complete")
models.create_task(conn, "P1-001", "p1", "Task") # execution_mode=None
mode = models.get_effective_mode(conn, "p1", "P1-001")
assert mode == "auto_complete"
def test_get_effective_mode_project_review_overrides_default(conn):
"""Project execution_mode='review' + task без override → возвращает 'review'.
Сценарий: PM хотел auto_complete, но проект настроен на review человеком.
get_effective_mode должен вернуть project-level 'review'.
"""
models.create_project(conn, "p1", "P1", "/p1", execution_mode="review")
models.create_task(conn, "P1-001", "p1", "Task") # нет task-level override
mode = models.get_effective_mode(conn, "p1", "P1-001")
assert mode == "review"
# -- Tasks --
def test_create_and_get_task(conn):
@ -238,3 +287,46 @@ def test_cost_summary(conn):
def test_cost_summary_empty(conn):
models.create_project(conn, "p1", "P1", "/p1")
assert models.get_cost_summary(conn, days=7) == []
# -- add_decision_if_new --
def test_add_decision_if_new_adds_new_decision(conn):
models.create_project(conn, "p1", "P1", "/p1")
d = models.add_decision_if_new(conn, "p1", "gotcha", "Use WAL mode", "description")
assert d is not None
assert d["title"] == "Use WAL mode"
assert d["type"] == "gotcha"
def test_add_decision_if_new_skips_exact_duplicate(conn):
models.create_project(conn, "p1", "P1", "/p1")
models.add_decision(conn, "p1", "gotcha", "Use WAL mode", "desc1")
result = models.add_decision_if_new(conn, "p1", "gotcha", "Use WAL mode", "desc2")
assert result is None
# Existing decision not duplicated
assert len(models.get_decisions(conn, "p1")) == 1
def test_add_decision_if_new_skips_case_insensitive_duplicate(conn):
models.create_project(conn, "p1", "P1", "/p1")
models.add_decision(conn, "p1", "decision", "Use UUID for task IDs", "desc")
result = models.add_decision_if_new(conn, "p1", "decision", "use uuid for task ids", "other desc")
assert result is None
assert len(models.get_decisions(conn, "p1")) == 1
def test_add_decision_if_new_allows_same_title_different_type(conn):
models.create_project(conn, "p1", "P1", "/p1")
models.add_decision(conn, "p1", "gotcha", "SQLite WAL", "desc")
result = models.add_decision_if_new(conn, "p1", "convention", "SQLite WAL", "other desc")
assert result is not None
assert len(models.get_decisions(conn, "p1")) == 2
def test_add_decision_if_new_skips_whitespace_duplicate(conn):
models.create_project(conn, "p1", "P1", "/p1")
models.add_decision(conn, "p1", "convention", "Run tests after each change", "desc")
result = models.add_decision_if_new(conn, "p1", "convention", " Run tests after each change ", "desc2")
assert result is None
assert len(models.get_decisions(conn, "p1")) == 1

838
tests/test_runner.py
View file

@ -6,7 +6,10 @@ import pytest
from unittest.mock import patch, MagicMock
from core.db import init_db
from core import models
from agents.runner import run_agent, run_pipeline, run_audit, _try_parse_json
from agents.runner import (
run_agent, run_pipeline, run_audit, _try_parse_json, _run_learning_extraction,
_build_claude_env, _resolve_claude_cmd, _EXTRA_PATH_DIRS, _run_autocommit,
)
@pytest.fixture
@ -155,8 +158,9 @@ class TestRunAgent:
# ---------------------------------------------------------------------------
class TestRunPipeline:
@patch("agents.runner._run_autocommit") # gotcha #41: мокируем в тестах не о autocommit
@patch("agents.runner.subprocess.run")
def test_successful_pipeline(self, mock_run, conn):
def test_successful_pipeline(self, mock_run, mock_autocommit, conn):
mock_run.return_value = _mock_claude_success({"result": "done"})
steps = [
@ -298,13 +302,13 @@ class TestAutoMode:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_generates_followups(self, mock_run, mock_hooks, mock_followup, conn):
"""Auto mode должен вызывать generate_followups после task_auto_approved."""
"""Auto_complete mode должен вызывать generate_followups (последний шаг — tester)."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -334,15 +338,15 @@ class TestAutoMode:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_skips_followups_for_followup_tasks(self, mock_run, mock_hooks, mock_followup, conn):
"""Auto mode НЕ должен генерировать followups для followup-задач (предотвращение рекурсии)."""
"""Auto_complete mode НЕ должен генерировать followups для followup-задач (предотвращение рекурсии)."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
models.update_task(conn, "VDOL-001", brief={"source": "followup:VDOL-000"})
steps = [{"role": "debugger", "brief": "find"}]
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -352,13 +356,13 @@ class TestAutoMode:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_fires_task_done_event(self, mock_run, mock_hooks, mock_followup, conn):
"""Auto mode должен вызывать run_hooks с event='task_done' после task_auto_approved."""
"""Auto_complete mode должен вызывать run_hooks с event='task_done' (последний шаг — tester)."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -371,7 +375,7 @@ class TestAutoMode:
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_mode_resolves_pending_actions(self, mock_run, mock_hooks, mock_followup, mock_resolve, conn):
"""Auto mode должен авто-резолвить pending_actions из followup generation."""
"""Auto_complete mode должен авто-резолвить pending_actions (последний шаг — tester)."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
@ -380,8 +384,8 @@ class TestAutoMode:
mock_followup.return_value = {"created": [], "pending_actions": pending}
mock_resolve.return_value = [{"resolved": "rerun", "result": {}}]
models.update_project(conn, "vdol", execution_mode="auto")
steps = [{"role": "debugger", "brief": "find"}]
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
@ -393,10 +397,12 @@ class TestAutoMode:
# ---------------------------------------------------------------------------
class TestRetryOnPermissionError:
@patch("agents.runner._run_autocommit")
@patch("agents.runner._run_learning_extraction")
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_retry_on_permission_error_auto_mode(self, mock_run, mock_hooks, mock_followup, conn):
def test_retry_on_permission_error_auto_mode(self, mock_run, mock_hooks, mock_followup, mock_learn, mock_autocommit, conn):
"""Auto mode: retry при permission error должен срабатывать."""
permission_fail = _mock_claude_failure("permission denied: cannot write file")
retry_success = _mock_claude_success({"result": "fixed"})
@ -404,8 +410,9 @@ class TestRetryOnPermissionError:
mock_run.side_effect = [permission_fail, retry_success]
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
mock_learn.return_value = {"added": 0, "skipped": 0}
models.update_project(conn, "vdol", execution_mode="auto")
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "debugger", "brief": "find"}]
result = run_pipeline(conn, "VDOL-001", steps)
@ -472,12 +479,13 @@ class TestNonInteractive:
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("stdin") == subprocess.DEVNULL
@patch.dict("os.environ", {"KIN_AGENT_TIMEOUT": ""}, clear=False)
@patch("agents.runner.subprocess.run")
def test_noninteractive_uses_300s_timeout(self, mock_run, conn):
def test_noninteractive_uses_600s_timeout(self, mock_run, conn):
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol", noninteractive=True)
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 300
assert call_kwargs.get("timeout") == 600
@patch.dict("os.environ", {"KIN_NONINTERACTIVE": ""})
@patch("agents.runner.subprocess.run")
@ -504,7 +512,16 @@ class TestNonInteractive:
run_agent(conn, "debugger", "VDOL-001", "vdol", noninteractive=False)
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("stdin") == subprocess.DEVNULL
assert call_kwargs.get("timeout") == 300
assert call_kwargs.get("timeout") == 600
@patch.dict("os.environ", {"KIN_AGENT_TIMEOUT": "900"})
@patch("agents.runner.subprocess.run")
def test_custom_timeout_via_env_var(self, mock_run, conn):
"""KIN_AGENT_TIMEOUT overrides the default 600s timeout."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 900
@patch("agents.runner.subprocess.run")
def test_allow_write_adds_skip_permissions(self, mock_run, conn):
@ -751,3 +768,786 @@ class TestSilentFailedDiagnostics:
assert result["success"] is True
assert result.get("error") is None
# ---------------------------------------------------------------------------
# Auto-learning: _run_learning_extraction
# ---------------------------------------------------------------------------
class TestRunLearningExtraction:
@patch("agents.runner.subprocess.run")
def test_extracts_and_saves_decisions(self, mock_run, conn):
"""Успешный сценарий: learner возвращает JSON с decisions, они сохраняются в БД."""
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "title": "SQLite WAL mode needed", "description": "Without WAL concurrent reads fail", "tags": ["sqlite", "db"]},
{"type": "convention", "title": "Always run tests after change", "description": "Prevents regressions", "tags": ["testing"]},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [
{"role": "debugger", "raw_output": "Found issue with sqlite concurrent access"},
]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 2
assert result["skipped"] == 0
decisions = conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()
assert len(decisions) == 2
titles = {d["title"] for d in decisions}
assert "SQLite WAL mode needed" in titles
assert "Always run tests after change" in titles
@patch("agents.runner.subprocess.run")
def test_skips_duplicate_decisions(self, mock_run, conn):
"""Дедупликация: если decision с таким title+type уже есть, пропускается."""
from core import models as m
m.add_decision(conn, "vdol", "gotcha", "SQLite WAL mode needed", "existing desc")
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "title": "SQLite WAL mode needed", "description": "duplicate", "tags": []},
{"type": "convention", "title": "New convention here", "description": "new desc", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [{"role": "tester", "raw_output": "test output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 1
assert result["skipped"] == 1
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 2
@patch("agents.runner.subprocess.run")
def test_limits_to_5_decisions(self, mock_run, conn):
"""Learner не должен сохранять более 5 decisions даже если агент вернул больше."""
decisions_list = [
{"type": "decision", "title": f"Decision {i}", "description": f"desc {i}", "tags": []}
for i in range(8)
]
learner_output = json.dumps({"decisions": decisions_list})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [{"role": "architect", "raw_output": "long output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 5
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 5
@patch("agents.runner.subprocess.run")
def test_non_json_output_returns_error(self, mock_run, conn):
"""Если learner вернул не-JSON, функция возвращает error, не бросает исключение."""
mock_run.return_value = _mock_claude_success({"result": "plain text, not json"})
step_results = [{"role": "debugger", "raw_output": "output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 0
assert "error" in result
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 0
@patch("agents.runner.subprocess.run")
def test_decisions_linked_to_task(self, mock_run, conn):
"""Сохранённые decisions должны быть привязаны к task_id."""
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "title": "Important gotcha", "description": "desc", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
d = conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchone()
assert d["task_id"] == "VDOL-001"
@patch("agents.runner._run_learning_extraction")
@patch("agents.runner.subprocess.run")
def test_pipeline_triggers_learning_after_completion(self, mock_run, mock_learn, conn):
"""run_pipeline должен вызывать _run_learning_extraction после успешного завершения."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_learn.return_value = {"added": 1, "skipped": 0}
steps = [{"role": "debugger", "brief": "find bug"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
mock_learn.assert_called_once()
call_args = mock_learn.call_args[0]
assert call_args[1] == "VDOL-001" # task_id
assert call_args[2] == "vdol" # project_id
@patch("agents.runner._run_learning_extraction")
@patch("agents.runner.subprocess.run")
def test_learning_error_does_not_break_pipeline(self, mock_run, mock_learn, conn):
"""Если _run_learning_extraction бросает исключение, pipeline не падает."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_learn.side_effect = Exception("learning failed")
steps = [{"role": "debugger", "brief": "find bug"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
def test_pipeline_dry_run_skips_learning(self, conn):
"""Dry run не должен вызывать _run_learning_extraction."""
steps = [{"role": "debugger", "brief": "find bug"}]
result = run_pipeline(conn, "VDOL-001", steps, dry_run=True)
assert result["dry_run"] is True
# No decisions saved (dry run — no DB activity)
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 0
@patch("agents.runner.subprocess.run")
def test_empty_learner_output_returns_no_decisions(self, mock_run, conn):
"""Пустой stdout от learner (subprocess вернул "") — не бросает исключение, возвращает error."""
# Используем пустую строку как stdout (не dict), чтобы raw_output оказался пустым
mock_run.return_value = _mock_claude_success("")
step_results = [{"role": "debugger", "raw_output": "output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 0
assert "error" in result
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 0
@patch("agents.runner.subprocess.run")
def test_empty_decisions_list_returns_zero_counts(self, mock_run, conn):
"""Learner возвращает {"decisions": []} — added=0, skipped=0, без ошибки."""
mock_run.return_value = _mock_claude_success({"result": json.dumps({"decisions": []})})
step_results = [{"role": "debugger", "raw_output": "output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 0
assert result["skipped"] == 0
assert "error" not in result
@patch("agents.runner.subprocess.run")
def test_decision_missing_title_is_skipped(self, mock_run, conn):
"""Decision без title молча пропускается, не вызывает исключение."""
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "description": "no title here", "tags": []},
{"type": "convention", "title": "Valid decision", "description": "desc", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [{"role": "debugger", "raw_output": "output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 1
assert len(conn.execute("SELECT * FROM decisions WHERE project_id='vdol'").fetchall()) == 1
@patch("agents.runner.subprocess.run")
def test_decisions_field_not_list_returns_error(self, mock_run, conn):
"""Если поле decisions не является списком — возвращается error dict."""
mock_run.return_value = _mock_claude_success({"result": json.dumps({"decisions": "not a list"})})
step_results = [{"role": "debugger", "raw_output": "output"}]
result = _run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
assert result["added"] == 0
assert "error" in result
@patch("agents.runner.subprocess.run")
def test_logs_agent_run_to_db(self, mock_run, conn):
"""KIN-060: _run_learning_extraction должна писать запись в agent_logs."""
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "title": "Log test", "description": "desc", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
logs = conn.execute(
"SELECT * FROM agent_logs WHERE agent_role='learner' AND project_id='vdol'"
).fetchall()
assert len(logs) == 1
log = logs[0]
assert log["task_id"] == "VDOL-001"
assert log["action"] == "learn"
assert log["model"] == "sonnet"
@patch("agents.runner.subprocess.run")
def test_learner_cost_included_in_cost_summary(self, mock_run, conn):
"""KIN-060: get_cost_summary() включает затраты learner-агента."""
learner_output = json.dumps({"decisions": []})
mock_run.return_value = _mock_claude_success({
"result": learner_output,
"cost_usd": 0.042,
"usage": {"total_tokens": 3000},
})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
costs = models.get_cost_summary(conn, days=1)
assert len(costs) == 1
assert costs[0]["project_id"] == "vdol"
assert costs[0]["total_cost_usd"] == pytest.approx(0.042)
assert costs[0]["total_tokens"] == 3000
# -----------------------------------------------------------------------
# KIN-061: Regression — валидация поля type в decision
# -----------------------------------------------------------------------
@patch("agents.runner.subprocess.run")
def test_valid_type_gotcha_is_saved_as_is(self, mock_run, conn):
"""KIN-061: валидный тип 'gotcha' сохраняется без изменений."""
learner_output = json.dumps({
"decisions": [
{"type": "gotcha", "title": "Use WAL mode", "description": "Concurrent reads need WAL", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
result = _run_learning_extraction(conn, "VDOL-001", "vdol", [{"role": "debugger", "raw_output": "x"}])
assert result["added"] == 1
d = conn.execute("SELECT type FROM decisions WHERE project_id='vdol'").fetchone()
assert d["type"] == "gotcha"
@patch("agents.runner.subprocess.run")
def test_invalid_type_falls_back_to_decision(self, mock_run, conn):
"""KIN-061: невалидный тип 'unknown_type' заменяется на 'decision'."""
learner_output = json.dumps({
"decisions": [
{"type": "unknown_type", "title": "Some title", "description": "Some desc", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
result = _run_learning_extraction(conn, "VDOL-001", "vdol", [{"role": "debugger", "raw_output": "x"}])
assert result["added"] == 1
d = conn.execute("SELECT type FROM decisions WHERE project_id='vdol'").fetchone()
assert d["type"] == "decision"
@patch("agents.runner.subprocess.run")
def test_missing_type_falls_back_to_decision(self, mock_run, conn):
"""KIN-061: отсутствующий ключ 'type' в decision заменяется на 'decision'."""
learner_output = json.dumps({
"decisions": [
{"title": "No type key here", "description": "desc without type", "tags": []},
]
})
mock_run.return_value = _mock_claude_success({"result": learner_output})
result = _run_learning_extraction(conn, "VDOL-001", "vdol", [{"role": "debugger", "raw_output": "x"}])
assert result["added"] == 1
d = conn.execute("SELECT type FROM decisions WHERE project_id='vdol'").fetchone()
assert d["type"] == "decision"
# -----------------------------------------------------------------------
# KIN-062: KIN_LEARNER_TIMEOUT — отдельный таймаут для learner-агента
# -----------------------------------------------------------------------
@patch.dict("os.environ", {"KIN_LEARNER_TIMEOUT": ""}, clear=False)
@patch("agents.runner.subprocess.run")
def test_learner_uses_120s_default_timeout(self, mock_run, conn):
"""KIN-062: по умолчанию learner использует таймаут 120s (KIN_LEARNER_TIMEOUT не задан)."""
mock_run.return_value = _mock_claude_success({"result": json.dumps({"decisions": []})})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 120
@patch.dict("os.environ", {"KIN_LEARNER_TIMEOUT": "300"}, clear=False)
@patch("agents.runner.subprocess.run")
def test_learner_uses_custom_timeout_from_env(self, mock_run, conn):
"""KIN-062: KIN_LEARNER_TIMEOUT переопределяет дефолтный таймаут learner-агента."""
mock_run.return_value = _mock_claude_success({"result": json.dumps({"decisions": []})})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 300
@patch.dict("os.environ", {"KIN_LEARNER_TIMEOUT": "60", "KIN_AGENT_TIMEOUT": "900"}, clear=False)
@patch("agents.runner.subprocess.run")
def test_learner_timeout_independent_of_agent_timeout(self, mock_run, conn):
"""KIN-062: KIN_LEARNER_TIMEOUT не зависит от KIN_AGENT_TIMEOUT."""
mock_run.return_value = _mock_claude_success({"result": json.dumps({"decisions": []})})
step_results = [{"role": "debugger", "raw_output": "output"}]
_run_learning_extraction(conn, "VDOL-001", "vdol", step_results)
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 60
# ---------------------------------------------------------------------------
# KIN-056: Regression — web path timeout parity with CLI
# ---------------------------------------------------------------------------
class TestRegressionKIN056:
"""Регрессионные тесты KIN-056: агенты таймаутили через 300s из web, но не из CLI.
Причина: noninteractive режим использовал timeout=300s.
Web API всегда устанавливает KIN_NONINTERACTIVE=1, поэтому таймаут был 300s.
Фикс: единый timeout=600s независимо от noninteractive (переопределяется KIN_AGENT_TIMEOUT).
Каждый тест ПАДАЛ бы со старым кодом (timeout=300 для noninteractive)
и ПРОХОДИТ после фикса.
"""
@patch.dict("os.environ", {"KIN_NONINTERACTIVE": "1", "KIN_AGENT_TIMEOUT": ""})
@patch("agents.runner.subprocess.run")
def test_web_noninteractive_env_does_not_use_300s(self, mock_run, conn):
"""Web путь устанавливает KIN_NONINTERACTIVE=1. До фикса это давало timeout=300s."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") != 300, (
"Регрессия KIN-056: timeout не должен быть 300s в noninteractive режиме"
)
@patch.dict("os.environ", {"KIN_NONINTERACTIVE": "1", "KIN_AGENT_TIMEOUT": ""})
@patch("agents.runner.subprocess.run")
def test_web_noninteractive_timeout_is_600(self, mock_run, conn):
"""Web путь: KIN_NONINTERACTIVE=1 → timeout = 600s (не 300s)."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 600
@patch("agents.runner.subprocess.run")
def test_web_and_cli_paths_use_same_timeout(self, mock_run, conn):
"""Таймаут через web-путь (KIN_NONINTERACTIVE=1) == таймаут CLI (noninteractive=True)."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
# Web path: env var KIN_NONINTERACTIVE=1, noninteractive param not set
with patch.dict("os.environ", {"KIN_NONINTERACTIVE": "1", "KIN_AGENT_TIMEOUT": ""}):
run_agent(conn, "debugger", "VDOL-001", "vdol", noninteractive=False)
web_timeout = mock_run.call_args[1].get("timeout")
mock_run.reset_mock()
# CLI path: noninteractive=True, no env var
with patch.dict("os.environ", {"KIN_NONINTERACTIVE": "", "KIN_AGENT_TIMEOUT": ""}):
run_agent(conn, "debugger", "VDOL-001", "vdol", noninteractive=True)
cli_timeout = mock_run.call_args[1].get("timeout")
assert web_timeout == cli_timeout, (
f"Таймаут web ({web_timeout}s) != CLI ({cli_timeout}s) — регрессия KIN-056"
)
@patch.dict("os.environ", {"KIN_NONINTERACTIVE": "1", "KIN_AGENT_TIMEOUT": "900"})
@patch("agents.runner.subprocess.run")
def test_web_noninteractive_respects_kin_agent_timeout_override(self, mock_run, conn):
"""Web путь: KIN_AGENT_TIMEOUT переопределяет дефолтный таймаут даже при KIN_NONINTERACTIVE=1."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
assert call_kwargs.get("timeout") == 900
# ---------------------------------------------------------------------------
# KIN-057: claude CLI в PATH при запуске через launchctl
# ---------------------------------------------------------------------------
class TestClaudePath:
"""Регрессионные тесты KIN-057: launchctl-демоны могут не видеть claude в PATH."""
def test_build_claude_env_contains_extra_paths(self):
"""_build_claude_env должен добавить /opt/homebrew/bin и /usr/local/bin в PATH."""
env = _build_claude_env()
path_dirs = env["PATH"].split(":")
for extra_dir in _EXTRA_PATH_DIRS:
assert extra_dir in path_dirs, (
f"Регрессия KIN-057: {extra_dir} не найден в PATH, сгенерированном _build_claude_env"
)
def test_build_claude_env_no_duplicate_paths(self):
"""_build_claude_env не должен дублировать уже существующие пути."""
env = _build_claude_env()
path_dirs = env["PATH"].split(":")
seen = set()
for d in path_dirs:
assert d not in seen, f"Дублирующийся PATH entry: {d}"
seen.add(d)
def test_build_claude_env_preserves_existing_path(self):
"""_build_claude_env должен сохранять уже существующие пути."""
with patch.dict("os.environ", {"PATH": "/custom/bin:/usr/bin:/bin"}):
env = _build_claude_env()
path_dirs = env["PATH"].split(":")
assert "/custom/bin" in path_dirs
assert "/usr/bin" in path_dirs
def test_resolve_claude_cmd_returns_string(self):
"""_resolve_claude_cmd должен всегда возвращать строку."""
cmd = _resolve_claude_cmd()
assert isinstance(cmd, str)
assert len(cmd) > 0
def test_resolve_claude_cmd_fallback_when_not_found(self):
"""_resolve_claude_cmd должен вернуть 'claude' если CLI не найден в PATH."""
with patch("agents.runner.shutil.which", return_value=None):
cmd = _resolve_claude_cmd()
assert cmd == "claude"
def test_resolve_claude_cmd_returns_full_path_when_found(self):
"""_resolve_claude_cmd должен вернуть полный путь если claude найден."""
with patch("agents.runner.shutil.which", return_value="/opt/homebrew/bin/claude"):
cmd = _resolve_claude_cmd()
assert cmd == "/opt/homebrew/bin/claude"
@patch("agents.runner.subprocess.run")
def test_run_claude_passes_env_to_subprocess(self, mock_run, conn):
"""_run_claude должен передавать env= в subprocess.run (а не наследовать голый PATH)."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
assert "env" in call_kwargs, (
"Регрессия KIN-057: subprocess.run должен получать явный env с расширенным PATH"
)
assert call_kwargs["env"] is not None
@patch("agents.runner.subprocess.run")
def test_run_claude_env_has_homebrew_in_path(self, mock_run, conn):
"""env переданный в subprocess.run должен содержать /opt/homebrew/bin в PATH."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
run_agent(conn, "debugger", "VDOL-001", "vdol")
call_kwargs = mock_run.call_args[1]
env = call_kwargs.get("env", {})
assert "/opt/homebrew/bin" in env.get("PATH", ""), (
"Регрессия KIN-057: /opt/homebrew/bin не найден в env['PATH'] subprocess.run"
)
@patch("agents.runner.subprocess.run")
def test_file_not_found_returns_127(self, mock_run, conn):
"""Если claude не найден (FileNotFoundError), должен вернуться returncode 127."""
mock_run.side_effect = FileNotFoundError("claude not found")
result = run_agent(conn, "debugger", "VDOL-001", "vdol")
assert result["success"] is False
assert "not found" in (result.get("error") or "").lower()
@patch.dict("os.environ", {"PATH": ""})
def test_launchctl_empty_path_build_env_adds_extra_dirs(self):
"""Регрессия KIN-057: когда launchctl запускает с пустым PATH,
_build_claude_env должен добавить _EXTRA_PATH_DIRS чтобы claude был доступен.
Без фикса: os.environ["PATH"]="" → shutil.which("claude") → None → FileNotFoundError.
После фикса: _build_claude_env строит PATH с /opt/homebrew/bin и др.
"""
env = _build_claude_env()
path_dirs = env["PATH"].split(":")
# Явная проверка каждой критичной директории
for extra_dir in _EXTRA_PATH_DIRS:
assert extra_dir in path_dirs, (
f"KIN-057: при пустом os PATH директория {extra_dir} должна быть добавлена"
)
@patch.dict("os.environ", {"PATH": ""})
def test_launchctl_empty_path_shutil_which_fails_without_fix(self):
"""Воспроизводит сломанное поведение: при PATH='' shutil.which возвращает None.
Это точно то, что происходило до фикса — launchctl не видел claude.
Тест документирует, ПОЧЕМУ нужен _build_claude_env вместо прямого os.environ.
"""
import shutil
# Без фикса: поиск с пустым PATH не найдёт claude
result_without_fix = shutil.which("claude", path="")
assert result_without_fix is None, (
"Если этот assert упал — shutil.which нашёл claude в пустом PATH, "
"что невозможно. Ожидаем None — именно поэтому нужен _build_claude_env."
)
# С фиксом: _resolve_claude_cmd строит расширенный PATH и находит claude
# (или возвращает fallback "claude", но не бросает FileNotFoundError)
cmd = _resolve_claude_cmd()
assert isinstance(cmd, str) and len(cmd) > 0, (
"KIN-057: _resolve_claude_cmd должен возвращать строку даже при пустом os PATH"
)
# ---------------------------------------------------------------------------
# KIN-063: TestCompletionMode — auto_complete + last-step role check
# ---------------------------------------------------------------------------
class TestCompletionMode:
"""auto_complete mode срабатывает только если последний шаг — tester или reviewer."""
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_complete_with_tester_last_sets_done(self, mock_run, mock_hooks, mock_followup, conn):
"""auto_complete + последний шаг tester → status=done (Decision #29)."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "developer", "brief": "fix"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "done"
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_complete_with_reviewer_last_sets_done(self, mock_run, mock_hooks, mock_followup, conn):
"""auto_complete + последний шаг reviewer → status=done."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "developer", "brief": "fix"}, {"role": "reviewer", "brief": "review"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "done"
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_auto_complete_without_tester_last_sets_review(self, mock_run, mock_hooks, mock_followup, conn):
"""auto_complete + последний шаг НЕ tester/reviewer → status=review (Decision #29)."""
mock_run.return_value = _mock_claude_success({"result": "ok"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
models.update_project(conn, "vdol", execution_mode="auto_complete")
steps = [{"role": "developer", "brief": "fix"}, {"role": "debugger", "brief": "debug"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "review", (
"Регрессия KIN-063: auto_complete без tester/reviewer последним НЕ должен авто-завершать"
)
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_legacy_auto_mode_value_not_recognized(self, mock_run, mock_hooks, mock_followup, conn):
"""Регрессия: старое значение 'auto' больше не является валидным режимом.
После KIN-063 'auto' → 'auto_complete'. Если в DB осталось 'auto' (без миграции),
runner НЕ должен авто-завершать — это 'review'-ветка (безопасный fallback).
(Decision #29)
"""
mock_run.return_value = _mock_claude_success({"result": "ok"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
# Прямой SQL-апдейт, обходя validate_completion_mode, чтобы симулировать
# старую запись в БД без миграции
conn.execute("UPDATE projects SET execution_mode='auto' WHERE id='vdol'")
conn.commit()
steps = [{"role": "developer", "brief": "fix"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "review", (
"Регрессия: 'auto' (старый формат) не должен срабатывать как auto_complete"
)
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_review_mode_with_tester_last_keeps_task_in_review(self, mock_run, mock_hooks, mock_followup, conn):
"""review mode + последний шаг tester → task.status == 'review', НЕ done (ждёт ручного approve)."""
mock_run.return_value = _mock_claude_success({"result": "all tests pass"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
# Проект и задача остаются в дефолтном 'review' mode
steps = [{"role": "developer", "brief": "fix"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "review"
assert task["status"] != "done", (
"KIN-063: review mode не должен авто-завершать задачу даже если tester последний"
)
@patch("core.followup.generate_followups")
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_project_review_overrides_no_task_completion_mode(self, mock_run, mock_hooks, mock_followup, conn):
"""Project execution_mode='review' + задача без override → pipeline завершается в 'review'.
Сценарий: PM выбрал auto_complete, но проект настроен на 'review' (ручной override человека).
Задача не имеет task-level execution_mode, поэтому get_effective_mode возвращает project-level 'review'.
"""
mock_run.return_value = _mock_claude_success({"result": "ok"})
mock_hooks.return_value = []
mock_followup.return_value = {"created": [], "pending_actions": []}
# Проект явно в 'review', задача без execution_mode
models.update_project(conn, "vdol", execution_mode="review")
# task VDOL-001 создана без execution_mode (None) — fixture
steps = [{"role": "developer", "brief": "fix"}, {"role": "tester", "brief": "test"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
assert result["mode"] == "review"
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "review", (
"KIN-063: project-level 'review' должен применяться когда задача не имеет override"
)
# ---------------------------------------------------------------------------
# KIN-048: _run_autocommit — флаг, git path, env=
# ---------------------------------------------------------------------------
class TestAutocommit:
"""KIN-048: _run_autocommit — autocommit_enabled флаг, shutil.which, env= regression."""
def test_disabled_project_skips_subprocess(self, conn):
"""autocommit_enabled=0 (дефолт) → subprocess не вызывается."""
with patch("agents.runner.subprocess.run") as mock_run:
_run_autocommit(conn, "VDOL-001", "vdol")
mock_run.assert_not_called()
@patch("agents.runner.subprocess.run")
@patch("agents.runner.shutil.which")
def test_enabled_calls_git_add_and_commit(self, mock_which, mock_run, conn, tmp_path):
"""autocommit_enabled=1 → вызываются git add -A и git commit с task_id и title."""
mock_which.return_value = "/usr/bin/git"
mock_run.return_value = MagicMock(returncode=0)
models.update_project(conn, "vdol", autocommit_enabled=1, path=str(tmp_path))
_run_autocommit(conn, "VDOL-001", "vdol")
assert mock_run.call_count == 2
add_cmd = mock_run.call_args_list[0][0][0]
assert add_cmd == ["/usr/bin/git", "add", "-A"]
commit_cmd = mock_run.call_args_list[1][0][0]
assert commit_cmd[0] == "/usr/bin/git"
assert commit_cmd[1] == "commit"
assert "VDOL-001" in commit_cmd[-1]
assert "Fix bug" in commit_cmd[-1]
@patch("agents.runner.subprocess.run")
def test_nothing_to_commit_no_exception(self, mock_run, conn, tmp_path):
"""returncode=1 (nothing to commit) → исключение не бросается."""
mock_run.return_value = MagicMock(returncode=1)
models.update_project(conn, "vdol", autocommit_enabled=1, path=str(tmp_path))
_run_autocommit(conn, "VDOL-001", "vdol") # must not raise
@patch("agents.runner.subprocess.run")
def test_passes_env_to_subprocess(self, mock_run, conn, tmp_path):
"""Regression #33: env= должен передаваться в subprocess.run."""
mock_run.return_value = MagicMock(returncode=0)
models.update_project(conn, "vdol", autocommit_enabled=1, path=str(tmp_path))
_run_autocommit(conn, "VDOL-001", "vdol")
for call in mock_run.call_args_list:
kwargs = call[1]
assert "env" in kwargs, "Regression #33: subprocess.run должен получать env="
assert "/opt/homebrew/bin" in kwargs["env"].get("PATH", "")
@patch("agents.runner.subprocess.run")
@patch("agents.runner.shutil.which")
def test_resolves_git_via_shutil_which(self, mock_which, mock_run, conn, tmp_path):
"""Regression #32: git резолвится через shutil.which, а не hardcoded 'git'."""
mock_which.return_value = "/opt/homebrew/bin/git"
mock_run.return_value = MagicMock(returncode=0)
models.update_project(conn, "vdol", autocommit_enabled=1, path=str(tmp_path))
_run_autocommit(conn, "VDOL-001", "vdol")
git_which_calls = [c for c in mock_which.call_args_list if c[0][0] == "git"]
assert len(git_which_calls) > 0, "Regression #32: shutil.which должен вызываться для git"
first_cmd = mock_run.call_args_list[0][0][0]
assert first_cmd[0] == "/opt/homebrew/bin/git"
@patch("agents.runner.subprocess.run")
@patch("agents.runner.shutil.which")
def test_git_not_found_no_crash_logs_warning(self, mock_which, mock_run, conn, tmp_path):
"""shutil.which(git) → None → fallback 'git' → FileNotFoundError → no crash, WARNING logged."""
mock_which.return_value = None # git не найден в PATH
mock_run.side_effect = FileNotFoundError("git: command not found")
models.update_project(conn, "vdol", autocommit_enabled=1, path=str(tmp_path))
with patch("agents.runner._logger") as mock_logger:
_run_autocommit(conn, "VDOL-001", "vdol") # не должен бросать исключение
mock_logger.warning.assert_called_once()
@patch("agents.runner._run_autocommit")
@patch("agents.runner.subprocess.run")
def test_autocommit_not_called_on_failed_pipeline(self, mock_run, mock_autocommit, conn):
"""Pipeline failure → _run_autocommit must NOT be called (gotcha #41)."""
mock_run.return_value = _mock_claude_failure("compilation error")
steps = [{"role": "debugger", "brief": "find"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is False
mock_autocommit.assert_not_called()
# ---------------------------------------------------------------------------
# KIN-055: execution_mode='review' при переводе задачи в статус review
# ---------------------------------------------------------------------------
class TestReviewModeExecutionMode:
"""Регрессия KIN-055: execution_mode должен быть 'review', а не NULL после pipeline в review mode."""
def test_task_execution_mode_is_null_before_pipeline(self, conn):
"""Граничный случай: execution_mode IS NULL до запуска pipeline (задача только создана)."""
task = models.get_task(conn, "VDOL-001")
assert task["execution_mode"] is None, (
"Задача должна иметь NULL execution_mode до выполнения pipeline"
)
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_review_mode_sets_execution_mode_review(self, mock_run, mock_hooks, conn):
"""После pipeline в review mode task.execution_mode должно быть 'review', а не NULL."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
steps = [{"role": "debugger", "brief": "find bug"}]
result = run_pipeline(conn, "VDOL-001", steps)
assert result["success"] is True
task = models.get_task(conn, "VDOL-001")
assert task["status"] == "review"
# Регрессионная проверка KIN-055: execution_mode не должен быть NULL
assert task["execution_mode"] is not None, (
"Регрессия KIN-055: execution_mode не должен быть NULL после перевода задачи в статус review"
)
assert task["execution_mode"] == "review"
@patch("agents.runner.run_hooks")
@patch("agents.runner.subprocess.run")
def test_review_mode_execution_mode_persisted_in_db(self, mock_run, mock_hooks, conn):
"""execution_mode='review' должно сохраняться в SQLite напрямую, минуя ORM-слой."""
mock_run.return_value = _mock_claude_success({"result": "done"})
mock_hooks.return_value = []
steps = [{"role": "debugger", "brief": "find"}]
run_pipeline(conn, "VDOL-001", steps)
row = conn.execute(
"SELECT execution_mode FROM tasks WHERE id='VDOL-001'"
).fetchone()
assert row is not None
assert row["execution_mode"] == "review", (
"Регрессия KIN-055: execution_mode должен быть 'review' в SQLite после pipeline"
)

62
web/api.py
View file

@ -3,6 +3,8 @@ Kin Web API — FastAPI backend reading ~/.kin/kin.db via core.models.
Run: uvicorn web.api:app --reload --port 8420
"""
import logging
import shutil
import subprocess
import sys
from pathlib import Path
@ -18,6 +20,7 @@ from pydantic import BaseModel
from core.db import init_db
from core import models
from core.models import VALID_COMPLETION_MODES
from agents.bootstrap import (
detect_tech_stack, detect_modules, extract_decisions_from_claude_md,
find_vault_root, scan_obsidian, save_to_db,
@ -25,6 +28,62 @@ from agents.bootstrap import (
DB_PATH = Path.home() / ".kin" / "kin.db"
_logger = logging.getLogger("kin")
# ---------------------------------------------------------------------------
# Startup: verify claude CLI is available in PATH
# ---------------------------------------------------------------------------
def _check_claude_available() -> None:
"""Warn at startup if claude CLI cannot be found in PATH.
launchctl daemons run with a stripped environment and may not see
/opt/homebrew/bin where claude is typically installed.
See Decision #28.
"""
from agents.runner import _build_claude_env # avoid circular import at module level
env = _build_claude_env()
claude_path = shutil.which("claude", path=env["PATH"])
if claude_path:
_logger.info("claude CLI found: %s", claude_path)
else:
_logger.warning(
"WARNING: claude CLI not found in PATH (%s). "
"Agent pipelines will fail with returncode 127. "
"Fix: add /opt/homebrew/bin to EnvironmentVariables.PATH in "
"~/Library/LaunchAgents/com.kin.api.plist and reload with: "
"launchctl unload ~/Library/LaunchAgents/com.kin.api.plist && "
"launchctl load ~/Library/LaunchAgents/com.kin.api.plist",
env.get("PATH", ""),
)
def _check_git_available() -> None:
"""Warn at startup if git cannot be found in PATH.
launchctl daemons run with a stripped environment and may not see
git in the standard directories. See Decision #28.
"""
from agents.runner import _build_claude_env # avoid circular import at module level
env = _build_claude_env()
git_path = shutil.which("git", path=env["PATH"])
if git_path:
_logger.info("git found: %s", git_path)
else:
_logger.warning(
"WARNING: git not found in PATH (%s). "
"Autocommit will fail silently. "
"Fix: add git directory to EnvironmentVariables.PATH in "
"~/Library/LaunchAgents/com.kin.api.plist and reload with: "
"launchctl unload ~/Library/LaunchAgents/com.kin.api.plist && "
"launchctl load ~/Library/LaunchAgents/com.kin.api.plist",
env.get("PATH", ""),
)
_check_claude_available()
_check_git_available()
app = FastAPI(title="Kin API", version="0.1.0")
app.add_middleware(
@ -162,7 +221,7 @@ class TaskPatch(BaseModel):
VALID_STATUSES = set(models.VALID_TASK_STATUSES)
VALID_EXECUTION_MODES = {"auto", "review"}
VALID_EXECUTION_MODES = VALID_COMPLETION_MODES
@app.patch("/api/tasks/{task_id}")
@ -361,6 +420,7 @@ def run_task(task_id: str):
cmd,
cwd=str(kin_root),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
env=env,
)