Private
Public Access
0
0

refactor(src): Phase 10.2 batch 3 - project_manager + orchestrator_pm Result migration

project_manager.py (3 sites):
- get_all_tracks returns list[dict[str, Any]] where each dict now
  has an 'errors' field (list[ErrorInfo]) capturing per-track
  metadata recovery. The 3 SILENT_SWALLOW sites (state.from_dict,
  metadata.json, plan.md) now append to this list instead of
  silently passing.

orchestrator_pm.py (2 sites):
- get_track_history_summary returns Result[str]. The 2 SILENT_SWALLOW
  sites (metadata.json + spec.md reads) append to a scan_errors list
  that's threaded through the Result.

Tests updated to check result.ok and use result.data.
This commit is contained in:
2026-06-17 22:33:57 -04:00
parent a7d8e2adfd
commit 89ce7ad770
3 changed files with 38 additions and 25 deletions
+9 -7
View File
@@ -8,14 +8,16 @@ from src import ai_client
from src import mma_prompts
from src import paths
from src import summarize
from src.result_types import Result, ErrorInfo, ErrorKind
def get_track_history_summary() -> str:
def get_track_history_summary() -> Result[str]:
"""
Scans conductor/archive/ and conductor/tracks/ to build a summary of past work.
[C: tests/test_orchestrator_pm_history.py:TestOrchestratorPMHistory.test_get_track_history_summary, tests/test_orchestrator_pm_history.py:TestOrchestratorPMHistory.test_get_track_history_summary_missing_files]
"""
summary_parts = []
scan_errors: list[ErrorInfo] = []
archive_path = paths.get_archive_dir()
tracks_path = paths.get_tracks_dir()
paths_to_scan = []
@@ -34,8 +36,8 @@ def get_track_history_summary() -> str:
meta = json.load(f)
title = meta.get("title", title)
status = meta.get("status", status)
except (OSError, json.JSONDecodeError, UnicodeDecodeError):
pass
except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
scan_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"orchestrator_pm.get_track_history_summary[{track_dir.name}].metadata", original=e))
if spec_file.exists():
try:
with open(spec_file, "r", encoding="utf-8") as f:
@@ -46,12 +48,12 @@ def get_track_history_summary() -> str:
else:
# Just take a snippet of the beginning
overview = content[:200] + "..."
except (OSError, UnicodeDecodeError):
pass
except (OSError, UnicodeDecodeError) as e:
scan_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"orchestrator_pm.get_track_history_summary[{track_dir.name}].spec", original=e))
summary_parts.append(f"Track: {title}\nStatus: {status}\nOverview: {overview}\n---")
if not summary_parts:
return "No previous tracks found."
return "\n".join(summary_parts)
return Result(data="No previous tracks found.", errors=scan_errors)
return Result(data="\n".join(summary_parts), errors=scan_errors)
def generate_tracks(user_request: str, project_config: dict[str, Any], file_items: list[dict[str, Any]], history_summary: Optional[str] = None) -> list[dict[str, Any]]:
"""
+16 -9
View File
@@ -332,16 +332,22 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
and 'progress' (0.0 to 1.0).
Handles missing or malformed metadata.json or state.toml by falling back
to available info or defaults.
Each returned dict includes an 'errors' list (list[ErrorInfo]) for any
per-track metadata recovery that occurred. Callers can ignore the errors
field for display purposes; the metadata is best-effort.
[C: tests/test_project_manager_tracks.py:test_get_all_tracks_empty, tests/test_project_manager_tracks.py:test_get_all_tracks_malformed, tests/test_project_manager_tracks.py:test_get_all_tracks_with_metadata_json, tests/test_project_manager_tracks.py:test_get_all_tracks_with_state, tests/test_project_paths.py:test_get_all_tracks_project_specific]
"""
tracks_dir = paths.get_tracks_dir(project_path=str(base_dir))
if not tracks_dir.exists(): return []
from src.result_types import ErrorInfo, ErrorKind
results: list[dict[str, Any]] = []
for entry in tracks_dir.iterdir():
if not entry.is_dir(): continue
track_id = entry.name
track_errors: list[dict[str, Any]] = []
track_info: dict[str, Any] = {
"id": track_id,
"title": track_id,
@@ -351,7 +357,7 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
"progress": 0.0
}
state_found = False
try:
state = load_track_state(track_id, base_dir)
if state:
@@ -363,8 +369,8 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
track_info["total"] = progress["total"]
track_info["progress"] = progress["percentage"] / 100.0
state_found = True
except (OSError, AttributeError, KeyError, TypeError):
pass
except (OSError, AttributeError, KeyError, TypeError) as e:
track_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"project_manager.get_all_tracks[{track_id}].state", original=e))
if not state_found:
metadata_file = entry / "metadata.json"
@@ -375,8 +381,8 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
track_info["id"] = data.get("id", data.get("track_id", track_id))
track_info["title"] = data.get("title", data.get("name", data.get("description", track_id)))
track_info["status"] = data.get("status", "unknown")
except (OSError, json.JSONDecodeError, UnicodeDecodeError):
pass
except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
track_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"project_manager.get_all_tracks[{track_id}].metadata", original=e))
if track_info["total"] == 0:
plan_file = entry / "plan.md"
@@ -390,9 +396,10 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
track_info["complete"] = len(completed_tasks)
if track_info["total"] > 0:
track_info["progress"] = float(track_info["complete"]) / track_info["total"]
except (OSError, UnicodeDecodeError, re.error):
pass
except (OSError, UnicodeDecodeError, re.error) as e:
track_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"project_manager.get_all_tracks[{track_id}].plan", original=e))
track_info["errors"] = track_errors
results.append(track_info)
return results
+13 -9
View File
@@ -37,12 +37,14 @@ class TestOrchestratorPMHistory(unittest.TestCase):
self.create_track(self.archive_dir, "track_001", "Initial Setup", "completed", "Setting up the project structure.")
self.create_track(self.tracks_dir, "track_002", "Feature A", "in_progress", "Implementing Feature A.")
summary = orchestrator_pm.get_track_history_summary()
self.assertIn("Initial Setup", summary)
self.assertIn("completed", summary)
self.assertIn("Setting up the project structure.", summary)
self.assertIn("Feature A", summary)
self.assertIn("in_progress", summary)
self.assertIn("Implementing Feature A.", summary)
self.assertTrue(summary.ok, f"get_track_history_summary failed: {summary.errors}")
body = summary.data
self.assertIn("Initial Setup", body)
self.assertIn("completed", body)
self.assertIn("Setting up the project structure.", body)
self.assertIn("Feature A", body)
self.assertIn("in_progress", body)
self.assertIn("Implementing Feature A.", body)
@patch('src.paths.get_archive_dir')
@patch('src.paths.get_tracks_dir')
@@ -54,9 +56,11 @@ class TestOrchestratorPMHistory(unittest.TestCase):
with open(track_path / "metadata.json", "w") as f:
json.dump({"title": "Missing Spec", "status": "pending"}, f)
summary = orchestrator_pm.get_track_history_summary()
self.assertIn("Missing Spec", summary)
self.assertIn("pending", summary)
self.assertIn("No overview available", summary)
self.assertTrue(summary.ok, f"get_track_history_summary failed: {summary.errors}")
body = summary.data
self.assertIn("Missing Spec", body)
self.assertIn("pending", body)
self.assertIn("No overview available", body)
@patch('src.orchestrator_pm.summarize.build_summary_markdown')
@patch('src.ai_client.send')