Implement core services for MVP-1 Personal Tracker CLI

Services implemented:
- session_service: Active session management (get/set/clear, validation)
- project_service: Project CRUD and directory structure
- note_service: Note handling and consolidation
- heuristics_service: Rule-based suggestions
- summary_service: Heuristic summary generation
This commit is contained in:
2026-03-23 08:54:09 -03:00
parent 4547c492da
commit b0c65a00a2
13 changed files with 757 additions and 0 deletions

709
tracker/cli/commands.py Normal file
View File

@@ -0,0 +1,709 @@
"""CLI commands implementation."""
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional
import typer
from tracker.models import Project, Session, NoteType, Change
from tracker.services import (
create_project,
get_project,
update_project,
list_projects,
ensure_project_structure,
get_active_session,
set_active_session,
clear_active_session,
validate_no_other_active_session,
add_note,
consolidate_notes,
suggest_next_steps,
generate_summary,
)
from tracker.storage import FileStorage, MarkdownReader, MarkdownWriter
# Initialize storage and markdown utilities
storage = FileStorage(projects_root=Path("projects"))
markdown_reader = MarkdownReader()
markdown_writer = MarkdownWriter()
# =============================================================================
# init-project command
# =============================================================================
init_project = typer.Typer(help="Create a new project with standard structure.")
@init_project.command("init-project")
def cmd_init_project(
name: str = typer.Argument(..., help="Project name"),
type: str = typer.Option("misc", help="Project type (code, homelab, automation, agent, research, misc)"),
tags: str = typer.Option("", help="Comma-separated tags"),
repo_path: Optional[str] = typer.Option(None, help="Path to git repository"),
description: str = typer.Option("", help="Project description"),
) -> None:
"""Create a new project with standard directory structure and files."""
from tracker.utils.slug import generate_slug
slug = generate_slug(name)
tag_list = [t.strip() for t in tags.split(",") if t.strip()]
# Create project using service
project = create_project(
name=name,
slug=slug,
description=description,
type=type,
tags=tag_list,
repo_path=Path(repo_path) if repo_path else None,
)
# Ensure directory structure
ensure_project_structure(slug)
# Generate files from templates
readme_content = _generate_readme(project)
log_content = "# Log\n\n_Project activity log_\n\n"
changelog_content = "# Changelog\n\n_Project changes_\n\n"
tasks_content = _generate_tasks_template()
meta_content = _generate_meta(project)
# Write files
storage.write_file(slug, "README.md", readme_content)
storage.write_file(slug, "LOG.md", log_content)
storage.write_file(slug, "CHANGELOG.md", changelog_content)
storage.write_file(slug, "TASKS.md", tasks_content)
storage.write_file(slug, "meta/project.yaml", meta_content)
typer.echo(f"Created project '{name}' with slug '{slug}'")
typer.echo(f"Location: {storage._project_path(slug)}")
# =============================================================================
# list command
# =============================================================================
list_projects = typer.Typer(help="List all projects.")
@list_projects.command("list")
def cmd_list_projects() -> None:
"""Show all projects with their status, last session, and next steps."""
projects = list_projects()
if not projects:
typer.echo("No projects found. Create one with: tracker init-project <name>")
return
typer.echo(f"\n{'Name':<25} {'Slug':<20} {'Status':<10} {'Last Session':<20} {'Next Step'}")
typer.echo("-" * 100)
for project in projects:
last_session = "Never"
next_step = ""
# Try to get last session info from LOG.md
log_content = storage.read_log(project.slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
if parsed.get("date_range"):
last_session = parsed["date_range"].split("")[0][:19]
if parsed.get("next_steps"):
next_step = parsed["next_steps"][0][:40]
typer.echo(
f"{project.name:<25} {project.slug:<20} {project.status:<10} "
f"{last_session:<20} {next_step}"
)
typer.echo(f"\nTotal: {len(projects)} project(s)")
# =============================================================================
# show command
# =============================================================================
show_project = typer.Typer(help="Show project details.")
@show_project.command("show")
def cmd_show_project(slug: str = typer.Argument(..., help="Project slug")) -> None:
"""Show detailed project information including status, context, last summary, blockers, and next steps."""
# Load project
project_dict = storage.read_project_meta(slug)
if not project_dict:
typer.echo(f"Error: Project '{slug}' not found.", err=True)
raise typer.Exit(code=1)
project = Project(**project_dict)
typer.echo(f"\n{'='*60}")
typer.echo(f"Project: {project.name}")
typer.echo(f"{'='*60}")
# Basic info
typer.echo(f"\nSlug: {project.slug}")
typer.echo(f"Type: {project.type}")
typer.echo(f"Status: {project.status}")
typer.echo(f"Description: {project.description or 'N/A'}")
if project.tags:
typer.echo(f"Tags: {', '.join(project.tags)}")
# Read LOG.md for context
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
# Last summary
if parsed.get("summary"):
typer.echo(f"\n--- Last Summary ---")
typer.echo(parsed["summary"][:300] + ("..." if len(parsed.get("summary", "")) > 300 else ""))
# Blockers
if parsed.get("blockers"):
typer.echo(f"\n--- Blockers ---")
for blocker in parsed["blockers"]:
typer.echo(f" - {blocker}")
# Next steps
if parsed.get("next_steps"):
typer.echo(f"\n--- Next Steps ---")
for step in parsed["next_steps"]:
typer.echo(f" - {step}")
# Last activity
if parsed.get("date_range"):
typer.echo(f"\n--- Last Activity ---")
typer.echo(f" {parsed['date_range']}")
typer.echo(f"\nLocation: {storage._project_path(slug)}")
# =============================================================================
# start command
# =============================================================================
start_session = typer.Typer(help="Start a work session.")
@start_session.command("start")
def cmd_start_session(
slug: str = typer.Argument(..., help="Project slug"),
objective: Optional[str] = typer.Option(None, help="Session objective"),
) -> None:
"""Start a new work session for the project.
Validates that no other session is active, creates an active session,
and shows recent context from the project.
"""
# Check project exists
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
# Validate no other active session
if not validate_no_other_active_session(slug):
active = get_active_session()
typer.echo(
f"Error: There is already an active session for project '{active.project_slug}'.",
err=True,
)
typer.echo(f"Stop that session first with: tracker stop {active.project_slug}")
raise typer.Exit(code=1)
# Create session
session = Session(
id=str(uuid.uuid4()),
project_slug=slug,
started_at=datetime.now(),
objective=objective or "",
)
# Save active session
set_active_session(session)
typer.echo(f"Started session for project '{slug}'")
if objective:
typer.echo(f"Objective: {objective}")
# Show recent context from LOG.md
typer.echo("\n--- Recent Context ---")
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
if parsed.get("work_done"):
typer.echo("Recent work:")
for item in parsed["work_done"][:3]:
typer.echo(f" - {item[:60]}")
if parsed.get("next_steps"):
typer.echo("Next steps:")
for item in parsed["next_steps"][:3]:
typer.echo(f" - {item}")
else:
typer.echo("No previous sessions found.")
# =============================================================================
# note command
# =============================================================================
add_note_cmd = typer.Typer(help="Add a note to the active session.")
@add_note_cmd.command("note")
def cmd_add_note(
text: str = typer.Argument(..., help="Note text"),
type: str = typer.Option("work", help="Note type (work, change, blocker, decision, idea, reference)"),
) -> None:
"""Add a note to the currently active session.
Note types:
- work: Work performed
- change: A change made
- blocker: Something blocking progress
- decision: A decision made
- idea: An idea
- reference: A reference or link
"""
# Get active session
session_data = storage.read_active_session()
if not session_data:
typer.echo("Error: No active session. Start one with: tracker start <slug>", err=True)
raise typer.Exit(code=1)
# Reconstruct session
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
# Add note
try:
note = add_note(session, type, text)
# Update active session file directly
storage.write_active_session(session.model_dump(mode="json"))
typer.echo(f"Added [{note['type']}] note: {text[:50]}{'...' if len(text) > 50 else ''}")
except ValueError as e:
typer.echo(f"Error: {e}", err=True)
raise typer.Exit(code=1)
# =============================================================================
# stop command
# =============================================================================
stop_session = typer.Typer(help="Stop the current session.")
@stop_session.command("stop")
def cmd_stop_session(
slug: str = typer.Argument(..., help="Project slug"),
add_to_changelog: bool = typer.Option(False, "--changelog", help="Add session summary to CHANGELOG.md"),
) -> None:
"""Stop the current session and generate summary.
Calculates duration, consolidates notes, generates summary,
suggests next steps, creates session file, updates LOG.md,
and optionally updates CHANGELOG.md.
"""
# Get active session
session_data = storage.read_active_session()
if not session_data:
typer.echo("Error: No active session to stop.", err=True)
raise typer.Exit(code=1)
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
# Verify it matches the slug
if session.project_slug != slug:
typer.echo(
f"Error: Active session is for project '{session.project_slug}', not '{slug}'.",
err=True,
)
raise typer.Exit(code=1)
# End session
session.ended_at = datetime.now()
session.duration_minutes = int((session.ended_at - session.started_at).total_seconds() / 60)
# Consolidate notes
consolidated = consolidate_notes(session.raw_notes)
session.work_done = consolidated["work_done"]
session.changes = consolidated["changes"]
session.decisions = consolidated["decisions"]
session.blockers = consolidated["blockers"]
session.references = consolidated["references"]
# Generate summary
session.summary = generate_summary(session)
# Get project for heuristics
project_dict = storage.read_project_meta(slug)
project = Project(**project_dict) if project_dict else None
# Suggest next steps
if project:
session.next_steps = suggest_next_steps(session, project)
else:
session.next_steps = suggest_next_steps(session, session)
# Save session file
storage.write_session_file(session)
# Format and append to LOG.md
log_entry = markdown_writer.format_log_entry(session, session.summary)
storage.append_to_log(slug, log_entry)
# Update README AUTOGEN sections
_update_readme_autogen(slug, session)
# Optionally add to changelog
if add_to_changelog and session.changes:
change = Change(
date=datetime.now().date(),
type="code",
title=f"Session: {session.objective or 'Work session'}",
impact=f"{session.duration_minutes} min - {len(session.work_done)} tasks completed",
)
changelog_entry = f"\n- **{change.date}** [{change.type}] {change.title}: {change.impact}"
storage.append_to_changelog(slug, changelog_entry)
# Clear active session
storage.delete_active_session()
# Print summary
typer.echo(f"\nSession completed for '{slug}'")
typer.echo(f"Duration: {session.duration_minutes} minutes")
typer.echo(f"\nSummary:\n{session.summary}")
if session.next_steps:
typer.echo("\nSuggested next steps:")
for step in session.next_steps:
typer.echo(f" - {step}")
# =============================================================================
# change command
# =============================================================================
add_change = typer.Typer(help="Add a change entry to CHANGELOG.md.")
@add_change.command("change")
def cmd_add_change(
slug: str = typer.Argument(..., help="Project slug"),
type: str = typer.Option("code", help="Change type (code, infra, config, docs, automation, decision)"),
title: str = typer.Option(..., help="Change title"),
impact: str = typer.Option("", help="Impact description"),
) -> None:
"""Add a manual entry to the project's CHANGELOG.md."""
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
change = Change(
date=datetime.now().date(),
type=type,
title=title,
impact=impact,
)
entry = f"- **{change.date}** [{change.type}] {change.title}: {change.impact}"
storage.append_to_changelog(slug, entry)
typer.echo(f"Added change to CHANGELOG.md:")
typer.echo(f" {entry}")
# =============================================================================
# next command
# =============================================================================
suggest_next = typer.Typer(help="Suggest next steps for a project.")
@suggest_next.command("next")
def cmd_suggest_next(slug: str = typer.Argument(..., help="Project slug")) -> None:
"""Suggest next steps based on project history and heuristics.
Uses simple rules:
- If there are open blockers, suggest unblocking
- If there are unvalidated changes, suggest validation
- If work is incomplete, suggest closing open threads
- If no progress, suggest redefining objective
"""
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
# Get active session if exists
session_data = storage.read_active_session()
if session_data:
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
else:
# Create a minimal session from project context
session = Session(
id="",
project_slug=slug,
started_at=datetime.now(),
)
# Load recent work from LOG
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
session.work_done = parsed.get("work_done", [])
session.changes = parsed.get("changes", [])
session.blockers = parsed.get("blockers", [])
session.decisions = parsed.get("decisions", [])
# Get project
project_dict = storage.read_project_meta(slug)
project = Project(**project_dict) if project_dict else None
# Get suggestions
suggestions = suggest_next_steps(session, project or Project(
id="",
name="",
slug=slug,
description="",
type="misc",
status="active",
tags=[],
root_path="",
created_at=datetime.now(),
updated_at=datetime.now(),
))
if suggestions:
typer.echo(f"\nSuggested next steps for '{slug}':")
for i, suggestion in enumerate(suggestions, 1):
typer.echo(f" {i}. {suggestion}")
else:
typer.echo(f"\nNo specific suggestions for '{slug}'.")
typer.echo("Consider defining new objectives or reviewing the project status.")
# =============================================================================
# review command
# =============================================================================
review = typer.Typer(help="Review all projects.")
@review.command("review")
def cmd_review() -> None:
"""Show an overview of all projects.
Displays:
- Active projects
- Recent sessions
- Open blockers
- Projects without recent activity
"""
projects = list_projects()
if not projects:
typer.echo("No projects to review.")
return
typer.echo("\n" + "=" * 60)
typer.echo("PROJECT REVIEW")
typer.echo("=" * 60)
# Categorize projects
active_projects = [p for p in projects if p.status == "active"]
blocked_projects = [p for p in projects if p.status == "blocked"]
other_projects = [p for p in projects if p.status not in ("active", "blocked")]
# Active projects
if active_projects:
typer.echo(f"\n--- Active Projects ({len(active_projects)}) ---")
for p in active_projects:
typer.echo(f" * {p.name} ({p.slug})")
else:
typer.echo("\n--- No Active Projects ---")
# Blocked projects
if blocked_projects:
typer.echo(f"\n--- Blocked Projects ({len(blocked_projects)}) ---")
for p in blocked_projects:
typer.echo(f" ! {p.name} ({p.slug})")
# Check for blockers in LOG
log_content = storage.read_log(p.slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
for blocker in parsed.get("blockers", []):
typer.echo(f" - {blocker}")
# Recent sessions
typer.echo(f"\n--- Recent Sessions ---")
has_recent = False
for p in projects:
sessions = storage.get_recent_sessions(p.slug, limit=1)
if sessions:
recent = sessions[0]
typer.echo(f" {p.name}: {recent.started_at.strftime('%Y-%m-%d %H:%M')} ({recent.duration_minutes} min)")
has_recent = True
if not has_recent:
typer.echo(" No recent sessions")
# Check for stale projects (no activity in 7 days)
typer.echo(f"\n--- Projects Without Recent Activity ---")
stale_threshold = datetime.now().timestamp() - (7 * 24 * 60 * 60)
stale_found = False
for p in projects:
if p.last_session_at:
if p.last_session_at.timestamp() < stale_threshold:
days = int((datetime.now() - p.last_session_at).days)
typer.echo(f" {p.name}: {days} days since last session")
stale_found = True
else:
# Check LOG for any sessions
log_content = storage.read_log(p.slug)
if not log_content:
typer.echo(f" {p.name}: Never had a session")
stale_found = True
if not stale_found:
typer.echo(" All projects have recent activity")
typer.echo("\n" + "=" * 60)
# =============================================================================
# Helper functions
# =============================================================================
def _generate_readme(project: Project) -> str:
"""Generate README.md content for a new project."""
return f"""# {project.name}
{project.description or '_No description_'}
## Objective
_TODO: Define objective_
## Status
**Current Status:** {project.status}
<!-- AUTOGEN:STATUS_START -->
Status: {project.status}
<!-- AUTOGEN:STATUS_END -->
## Context
_Current context and background_
## Stack / Tools
- _Tool 1_
- _Tool 2_
## Architecture
_Brief architecture description_
## Technical Decisions
_No decisions recorded yet_
## Risks / Blockers
_No blockers_
<!-- AUTOGEN:NEXT_STEPS_START -->
<!-- AUTOGEN:NEXT_STEPS_END -->
## Recent Sessions
<!-- AUTOGEN:SESSIONS_START -->
<!-- AUTOGEN:SESSIONS_END -->
_Last updated: {datetime.now().strftime('%Y-%m-%d')}_
"""
def _generate_tasks_template() -> str:
"""Generate TASKS.md content for a new project."""
return """# Tasks
## Inbox
-
## Next
-
## In Progress
-
## Blocked
-
## Waiting
-
## Done
-
"""
def _generate_meta(project: Project) -> str:
"""Generate meta/project.yaml content."""
import yaml
data = {
"id": project.id,
"name": project.name,
"slug": project.slug,
"description": project.description,
"type": project.type,
"status": project.status,
"tags": project.tags,
"root_path": str(project.root_path),
"repo_path": str(project.repo_path) if project.repo_path else None,
"created_at": project.created_at.isoformat(),
"updated_at": project.updated_at.isoformat(),
"last_session_at": None,
}
return yaml.dump(data, default_flow_style=False, sort_keys=False)
def _update_readme_autogen(slug: str, session: Session) -> None:
"""Update README.md AUTOGEN sections with session info."""
# Update sessions section
session_line = f"- {session.started_at.strftime('%Y-%m-%d %H:%M')} ({session.duration_minutes} min): {session.summary[:50]}..."
sessions_content = storage.extract_autogen_section(slug, "SESSIONS") or ""
sessions_content = sessions_content + f"\n{session_line}" if sessions_content else session_line
storage.update_readme_autogen(slug, "SESSIONS", sessions_content.strip())
# Update next steps section
if session.next_steps:
next_steps_content = "\n".join([f"- {step}" for step in session.next_steps])
existing_next = storage.extract_autogen_section(slug, "NEXT_STEPS") or ""
if existing_next:
# Keep existing and add new
next_steps_content = existing_next + "\n" + next_steps_content
storage.update_readme_autogen(slug, "NEXT_STEPS", next_steps_content.strip())
# Register all commands at module level for direct access
__all__ = [
"init_project",
"list_projects",
"show_project",
"start_session",
"add_note_cmd",
"stop_session",
"add_change",
"suggest_next",
"review",
]