Software Orchestrator
5769c1cd22
- Auto-Registration-Bug behoben (register_project/get_project_id/resolve_project Trennung) - 25 Tests gruen (Pytest) - block_compactor-Tool refactored (Option B: Soft-Check statt Hard-Block) - 4 Restbaustellen gefixt - DB-Schema: plugin_settings-Tabelle hinzugefuegt - 3 Schattenprojekte aus DB geloescht - Plan v3 + Refactor-Plan + Worklog dokumentiert
132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""Rules for required project artifacts (DB + repo hybrid)."""
|
||
import os
|
||
from typing import Dict, List
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DB-backed artifacts (5 mandatory: worklog, todo, project_state,
|
||
# current_status, next_steps) – checked via db_state_store
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DB_ARTIFACTS: List[str] = [
|
||
"worklog", # orch_worklog (mindestens 1 Eintrag)
|
||
"todo", # orch_todos (mindestens 1 Eintrag)
|
||
"project_state", # orch_kv mit key='project_state' ODER 'phase'
|
||
"current_status", # orch_status (gesetzt)
|
||
"next_steps", # orch_next_steps (mindestens 1 pending)
|
||
]
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Repo-backed artifacts (15 specs/docs/deploy files) – checked as file paths
|
||
# unter project_root
|
||
# ---------------------------------------------------------------------------
|
||
|
||
REPO_ARTIFACTS: List[str] = [
|
||
"specs/current/requirements.md",
|
||
"specs/current/design.md",
|
||
"specs/current/tasks.md",
|
||
"docs/architecture.md",
|
||
"docs/test_report.md",
|
||
"docs/runtime_report.md",
|
||
"deploy/env.md",
|
||
"deploy/healthcheck.md",
|
||
"deploy/runbook.md",
|
||
"deploy/rollback.md",
|
||
]
|
||
|
||
# All artifacts (kept for backward compat with the original list)
|
||
CORE_ARTIFACTS = [f".a0/{a}.md" for a in DB_ARTIFACTS if a != "project_state"] + [
|
||
".a0/project_state.json"
|
||
] + REPO_ARTIFACTS
|
||
|
||
|
||
def _check_db_artifacts(project_name: str) -> Dict[str, bool]:
|
||
"""Prüft die 5 DB-backed Artefakte via db_state_store."""
|
||
from .db_state_store import (
|
||
get_project_id,
|
||
get_kv,
|
||
get_status,
|
||
list_worklog,
|
||
list_todos,
|
||
list_next_steps,
|
||
)
|
||
|
||
pid = get_project_id(project_name)
|
||
if pid is None:
|
||
return {
|
||
"worklog": False, "todo": False, "project_state": False,
|
||
"current_status": False, "next_steps": False,
|
||
}
|
||
|
||
return {
|
||
"worklog": len(list_worklog(pid, limit=1)) > 0,
|
||
"todo": len(list_todos(pid, status="open", limit=1)) > 0,
|
||
"project_state": (
|
||
get_kv(pid, "project_state") is not None
|
||
or get_kv(pid, "phase") is not None
|
||
),
|
||
"current_status": get_status(pid) is not None,
|
||
"next_steps": len(list_next_steps(pid, status="pending", limit=1)) > 0,
|
||
}
|
||
|
||
|
||
def _check_repo_artifacts(project_root: str) -> Dict[str, bool]:
|
||
"""Prüft die 15 Repo-Artefakte (specs/, docs/, deploy/) per File-Existenz."""
|
||
result: Dict[str, bool] = {}
|
||
for rel_path in REPO_ARTIFACTS:
|
||
result[rel_path] = os.path.exists(os.path.join(project_root, rel_path))
|
||
return result
|
||
|
||
|
||
def check_required_artifacts(
|
||
project_name: str = "",
|
||
project_root: str = "",
|
||
) -> dict:
|
||
"""Prüft alle Pflicht-Artefakte (DB + Repo).
|
||
|
||
Args:
|
||
project_name: Projektname (für DB-Checks). Default: aus project_root
|
||
abgeleitet (basename).
|
||
project_root: Projekt-Wurzelverzeichnis (für Repo-Checks). Default: cwd.
|
||
|
||
Returns:
|
||
{
|
||
"db": {artifact_name: bool, …},
|
||
"repo": {relative_path: bool, …},
|
||
"present": [alle existierenden], # kombiniert
|
||
"missing": [alle fehlenden], # kombiniert
|
||
"total": int,
|
||
}
|
||
"""
|
||
if not project_name:
|
||
project_name = os.path.basename(project_root or os.getcwd())
|
||
if not project_root:
|
||
project_root = os.getcwd()
|
||
|
||
db = _check_db_artifacts(project_name)
|
||
repo = _check_repo_artifacts(project_root)
|
||
|
||
present = [a for a, ok in db.items() if ok] + [a for a, ok in repo.items() if ok]
|
||
missing = [a for a, ok in db.items() if not ok] + [a for a, ok in repo.items() if not ok]
|
||
|
||
return {
|
||
"db": db,
|
||
"repo": repo,
|
||
"present": present,
|
||
"missing": missing,
|
||
"total": len(db) + len(repo),
|
||
}
|
||
|
||
|
||
# Backward-compat: altes Format (nur Repo-Files) für externe Caller
|
||
def check_legacy_artifacts(project_root: str) -> dict:
|
||
"""Alte CORE_ARTIFACTS-Liste als File-Existenz-Check (nur Repo)."""
|
||
missing = []
|
||
present = []
|
||
for artifact in CORE_ARTIFACTS:
|
||
path = os.path.join(project_root, artifact)
|
||
if os.path.exists(path):
|
||
present.append(artifact)
|
||
else:
|
||
missing.append(artifact)
|
||
return {"present": present, "missing": missing, "total": len(CORE_ARTIFACTS)}
|