"""View commands for displaying plan information."""
import json
import sys
from pathlib import Path
import jsonschema
from plan_view.formatting import ICONS, bold, bold_cyan, bold_yellow, dim, green
from plan_view.io import load_schema
from plan_view.state import (
SPECIAL_PHASE_IDS,
find_phase,
find_task,
get_current_phase,
get_next_task,
task_to_dict,
)
# ============================================================================
# AI-Optimized Output Functions (Context-Efficient)
# ============================================================================
def _ai_task_line(task: dict, phase: dict | None = None) -> str:
"""Single-line compact task representation for AI agents."""
parts = [task["id"], task["status"], task["title"]]
if task.get("agent_type"):
parts.append(f"agent:{task['agent_type']}")
if task.get("skill"):
parts.append(f"skill:{task['skill']}")
if task.get("depends_on"):
parts.append(f"deps:{','.join(task['depends_on'])}")
if phase:
parts.append(f"phase:{phase['id']}")
return " | ".join(parts)
def _ai_phase_line(phase: dict) -> str:
"""Single-line compact phase representation for AI agents."""
progress = phase.get("progress", {})
return f"{phase['id']} {phase['name']} ({progress.get('completed', 0)}/{progress.get('total', 0)})"
[docs]
def cmd_ai_context(plan: dict) -> None:
"""Output minimal context for AI agents: next task + current phase + key info."""
lines = []
# Summary line
summary = plan.get("summary", {})
pct = summary.get("overall_progress", 0)
done = summary.get("completed_tasks", 0)
total = summary.get("total_tasks", 0)
lines.append(f"PROGRESS: {pct:.0f}% ({done}/{total})")
# Current phase
current = get_current_phase(plan)
if current:
lines.append(f"PHASE: {_ai_phase_line(current)}")
# Next task with full context
result = get_next_task(plan)
if result:
phase, task = result
lines.append(f"NEXT: {_ai_task_line(task, phase)}")
# Include files if present
if task.get("files"):
lines.append(f"FILES: {', '.join(task['files'])}")
# Include plan snippet if present
if task.get("plan"):
plan_preview = task["plan"][:200] + "..." if len(task["plan"]) > 200 else task["plan"]
lines.append(f"PLAN: {plan_preview}")
else:
lines.append("NEXT: none")
print("\n".join(lines))
[docs]
def cmd_ai_actionable(plan: dict) -> None:
"""Output only actionable tasks (pending with deps met) for AI agents."""
task_status = {t["id"]: t["status"] for p in plan.get("phases", []) for t in p.get("tasks", [])}
lines = []
for phase in plan.get("phases", []):
if phase["id"] in SPECIAL_PHASE_IDS or phase["status"] in ("completed", "skipped"):
continue
for task in phase.get("tasks", []):
if task["status"] in ("pending", "in_progress"):
deps = task.get("depends_on", [])
if all(task_status.get(dep) == "completed" for dep in deps):
lines.append(_ai_task_line(task, phase))
print("\n".join(lines) if lines else "none")
[docs]
def cmd_ai_files(plan: dict, task_id: str | None = None) -> None:
"""Output files for a task or current task."""
result = find_task(plan, task_id) if task_id else get_next_task(plan)
if not result:
print("none")
return
_, task = result
files = task.get("files", [])
print("\n".join(files) if files else "none")
HELP_TEXT = """\
View and edit plan.json for task tracking
View Commands:
(none) Show plan overview (completed phases hidden by default)
current, c Show current progress and next task
next, n Show next task to work on
phase, p Show current phase details
get, g ID Show a specific task or phase by ID
last, l [-a] Show recently completed tasks (-a for all)
future, f [-a] Show upcoming tasks (-a for all)
summary, s Show plan summary (pretty output, use --json for JSON)
table, t [PHASE] Show tasks in table format (optional: filter by phase)
bugs, b Show bugs phase with all tasks
deferred, d Show deferred phase with all tasks
ideas, i Show ideas phase with all tasks
validate, v Validate plan.json structure
AI-Optimized Commands (token-efficient output):
--ai Compact context: progress, phase, next task with files/plan
--ai actionable List only actionable tasks (deps met)
files [ID] Show files for task ID or next task
Edit Commands:
init NAME Create new plan.json
add-phase NAME Add a new phase
add-task PHASE TITLE Add a new task to a phase (--agent, --skill, --files)
set ID FIELD VALUE Set a task field (status, agent, skill, title, files, research, plan)
done ID Mark task as completed
start ID Mark task as in_progress
block ID Mark task as blocked
skip ID Mark task as skipped
defer ID|TITLE Move task to deferred, or add new deferred task
bug ID|TITLE Move task to bugs, or add new bug task
idea ID|TITLE Move task to ideas, or add new idea task
rm TYPE ID Remove a phase or task
Utilities:
dashboard Open interactive dashboard in browser
reconcile Fix data inconsistencies and validate
compact Remove tracking data from completed tasks
Options:
-f, --file FILE Path to plan.json (default: ./plan.json)
-a, --all Show all phases including completed (default view only)
--json Output as JSON (view commands only)
--ai AI-optimized compact output (context-efficient)
-q, --quiet Suppress output (edit commands only)
-d, --dry-run Show what would change without saving
-h, --help Show this help message
Configuration (for monorepos):
pv searches for config files walking up from cwd:
1. .pv.toml - dedicated config file
2. pyproject.toml - [tool.pv] section
Config format:
plan_file = "path/to/plan.json" # relative to config location
"""
[docs]
def cmd_overview(plan: dict, *, as_json: bool = False, show_all: bool = False) -> None:
"""Display full plan overview with all phases and tasks.
By default, completed phases are hidden. Use --all to show everything.
"""
if as_json:
print(json.dumps(plan, indent=2))
return
meta = plan.get("meta", {})
summary = plan.get("summary", {})
project = meta.get("project", "Unknown Project")
version = meta.get("version", "0.0.0")
total = summary.get("total_tasks", 0)
completed = summary.get("completed_tasks", 0)
pct = summary.get("overall_progress", 0)
print(f"\n{bold(f'📋 {project} v{version}')}")
print(f"Progress: {pct:.0f}% ({completed}/{total} tasks)\n")
# Filter phases based on show_all flag
all_phases = plan.get("phases", [])
phases_to_show = all_phases if show_all else [p for p in all_phases if p["status"] != "completed"]
# Show count of hidden completed phases
hidden_count = len(all_phases) - len(phases_to_show)
if hidden_count > 0:
print(dim(f"({hidden_count} completed phase{'s' if hidden_count != 1 else ''} hidden, use -a to show all)\n"))
for phase in phases_to_show:
progress = phase.get("progress", {})
phase_pct = progress.get("percentage", 0)
icon = ICONS.get(phase["status"], "❓")
phase_id = phase["id"]
phase_name = phase["name"]
phase_desc = phase["description"]
print(f"{icon} {bold(f'Phase {phase_id}: {phase_name}')} ({phase_pct:.0f}%)")
print(f" {phase_desc}\n")
for task in phase.get("tasks", []):
t_icon = ICONS.get(task["status"], "❓")
task_id = task["id"]
task_title = task["title"]
agent = task.get("agent_type") or "general"
skill = task.get("skill")
skill_str = f" [skill: {skill}]" if skill else ""
print(f" {t_icon} [{task_id}] {task_title} {dim(f'({agent})')}{dim(skill_str)}")
print()
[docs]
def cmd_current(plan: dict, *, as_json: bool = False) -> None:
"""Display completed phases summary, current phase, and next task."""
if as_json:
current = get_current_phase(plan)
result = get_next_task(plan)
output = {
"summary": plan.get("summary", {}),
"current_phase": current,
"next_task": task_to_dict(*result) if result else None,
}
print(json.dumps(output, indent=2))
return
meta = plan.get("meta", {})
summary = plan.get("summary", {})
project = meta.get("project", "Unknown Project")
version = meta.get("version", "0.0.0")
total = summary.get("total_tasks", 0)
completed = summary.get("completed_tasks", 0)
pct = summary.get("overall_progress", 0)
print(f"\n{bold(f'📋 {project} v{version}')}")
print(f"Progress: {pct:.0f}% ({completed}/{total} tasks)\n")
for phase in plan.get("phases", []):
if phase["status"] == "completed":
phase_id = phase["id"]
phase_name = phase["name"]
print(green(f"✅ Phase {phase_id}: {phase_name} (100%)"))
current = get_current_phase(plan)
if current:
progress = current.get("progress", {})
pct = progress.get("percentage", 0)
status_icon = "🔄" if current["status"] == "in_progress" else "⏳"
phase_id = current["id"]
phase_name = current["name"]
phase_desc = current["description"]
print(f"\n{status_icon} {bold_yellow(f'Phase {phase_id}: {phase_name} ({pct:.0f}%)')}")
print(f" {phase_desc}\n")
for task in current.get("tasks", []):
icon = ICONS.get(task["status"], "❓")
task_id = task["id"]
task_title = task["title"]
agent = task.get("agent_type") or "general"
skill = task.get("skill")
skill_str = f" [skill: {skill}]" if skill else ""
print(f" {icon} [{task_id}] {task_title} {dim(f'({agent})')}{dim(skill_str)}")
result = get_next_task(plan)
if result:
_, task = result
task_id = task["id"]
task_title = task["title"]
print(f"\n{bold('👉 Next:')} [{task_id}] {task_title}")
print()
[docs]
def cmd_next(plan: dict, *, as_json: bool = False) -> None:
"""Display the next task to work on."""
result = get_next_task(plan)
if not result:
if as_json:
print("null")
else:
print("No pending tasks found!")
return
phase, task = result
if as_json:
print(json.dumps(task_to_dict(phase, task), indent=2))
return
icon = ICONS.get(task["status"], "❓")
agent = task.get("agent_type") or "general-purpose"
skill = task.get("skill")
task_id = task["id"]
task_title = task["title"]
phase_name = phase["name"]
print(f"\n{bold('Next Task:')}")
print(f" {icon} [{task_id}] {task_title}")
print(f" {dim('Phase:')} {phase_name}")
print(f" {dim('Agent:')} {agent}")
if skill:
print(f" {dim('Skill:')} {skill}")
deps = task.get("depends_on", [])
if deps:
deps_str = ", ".join(deps)
print(f" {dim('Depends on:')} {deps_str}")
print()
[docs]
def cmd_phase(plan: dict, *, as_json: bool = False) -> None:
"""Display current phase details with all tasks and dependencies."""
phase = get_current_phase(plan)
if not phase:
if as_json:
print("null")
else:
print("No active phase found!")
return
if as_json:
print(json.dumps(phase, indent=2))
return
progress = phase.get("progress", {})
pct = progress.get("percentage", 0)
phase_id = phase["id"]
phase_name = phase["name"]
phase_desc = phase["description"]
completed = progress.get("completed", 0)
total = progress.get("total", 0)
print(f"\n{bold_cyan(f'Phase {phase_id}: {phase_name}')}")
print(f" {phase_desc}")
print(f" Progress: {pct:.0f}% ({completed}/{total} tasks)\n")
for task in phase.get("tasks", []):
icon = ICONS.get(task["status"], "❓")
task_id = task["id"]
task_title = task["title"]
agent = task.get("agent_type") or "general"
agent_str = f"({agent})" if task.get("agent_type") else ""
skill = task.get("skill")
skill_str = f" [skill: {skill}]" if skill else ""
deps = task.get("depends_on", [])
dep_str = f" [deps: {', '.join(deps)}]" if deps else ""
print(f" {icon} [{task_id}] {task_title} {dim(agent_str)}{dim(skill_str)}{dim(dep_str)}")
print()
[docs]
def cmd_get(plan: dict, task_id: str, *, as_json: bool = False) -> None:
"""Display a specific task or phase by ID."""
# Try to find as a task first
result = find_task(plan, task_id)
if result:
phase, task = result
if as_json:
print(json.dumps(task_to_dict(phase, task), indent=2))
return
icon = ICONS.get(task["status"], "❓")
agent = task.get("agent_type") or "general-purpose"
skill = task.get("skill")
tracking = task.get("tracking", {})
print(f"\n{bold(f'[{task_id}] {task["title"]}')}")
print(f" {dim('Status:')} {icon} {task['status']}")
print(f" {dim('Phase:')} {phase['name']}")
print(f" {dim('Agent:')} {agent}")
if skill:
print(f" {dim('Skill:')} {skill}")
deps = task.get("depends_on", [])
if deps:
print(f" {dim('Depends on:')} {', '.join(deps)}")
if tracking.get("started_at"):
print(f" {dim('Started:')} {tracking['started_at'][:10]}")
if tracking.get("completed_at"):
print(f" {dim('Completed:')} {tracking['completed_at'][:10]}")
if tracking.get("defer_reason"):
print(f" {dim('Defer reason:')} {tracking['defer_reason']}")
print()
return
# If not a task, try to find as a phase
phase = find_phase(plan, task_id)
if phase:
if as_json:
print(json.dumps(phase, indent=2))
return
progress = phase.get("progress", {})
pct = progress.get("percentage", 0)
phase_id = phase["id"]
phase_name = phase["name"]
phase_desc = phase["description"]
completed = progress.get("completed", 0)
total = progress.get("total", 0)
print(f"\n{bold_cyan(f'Phase {phase_id}: {phase_name}')}")
print(f" {phase_desc}")
print(f" Progress: {pct:.0f}% ({completed}/{total} tasks)\n")
for task in phase.get("tasks", []):
icon = ICONS.get(task["status"], "❓")
task_id_display = task["id"]
task_title = task["title"]
agent = task.get("agent_type") or "general"
agent_str = f"({agent})" if task.get("agent_type") else ""
skill = task.get("skill")
skill_str = f" [skill: {skill}]" if skill else ""
deps = task.get("depends_on", [])
dep_str = f" [deps: {', '.join(deps)}]" if deps else ""
print(f" {icon} [{task_id_display}] {task_title} {dim(agent_str)}{dim(skill_str)}{dim(dep_str)}")
print()
return
# Not found as either task or phase
if as_json:
print("null")
else:
print(f"Task or phase '{task_id}' not found!")
return
[docs]
def cmd_last(plan: dict, count: int | None = 5, *, as_json: bool = False) -> None:
"""Display recently completed tasks."""
completed_tasks = []
for phase in plan.get("phases", []):
for task in phase.get("tasks", []):
if task["status"] == "completed":
tracking = task.get("tracking", {})
completed_at = tracking.get("completed_at")
completed_tasks.append((phase, task, completed_at))
if not completed_tasks:
if as_json:
print("[]")
else:
print("No completed tasks found!")
return
# Sort by completion time (most recent first), tasks without timestamp go last
completed_tasks.sort(key=lambda x: x[2] or "", reverse=True)
if as_json:
output = [
{
"id": task["id"],
"title": task["title"],
"phase_id": phase["id"],
"phase_name": phase["name"],
"completed_at": completed_at,
"agent_type": task.get("agent_type"),
"skill": task.get("skill"),
}
for phase, task, completed_at in completed_tasks[:count]
]
print(json.dumps(output, indent=2))
return
print(f"\n{bold('Recently Completed:')}\n")
for phase, task, completed_at in completed_tasks[:count]:
task_id = task["id"]
task_title = task["title"]
phase_name = phase["name"]
time_str = completed_at[:10] if completed_at else "unknown"
print(f" ✅ [{task_id}] {task_title}")
print(f" {dim(f'{phase_name} · {time_str}')}")
print()
[docs]
def cmd_future(plan: dict, count: int | None = 5, *, as_json: bool = False) -> None:
"""Display upcoming tasks (pending/in_progress), actionable first."""
# Build task status lookup for dependency checks
task_status = {t["id"]: t["status"] for p in plan.get("phases", []) for t in p.get("tasks", [])}
future_tasks: list[tuple[dict, dict, bool]] = [] # (phase, task, is_actionable)
for phase in plan.get("phases", []):
# Skip special phases
if phase["id"] in SPECIAL_PHASE_IDS:
continue
# Skip completed/skipped phases
if phase["status"] in ("completed", "skipped"):
continue
for task in phase.get("tasks", []):
if task["status"] in ("pending", "in_progress", "blocked"):
# Check if actionable (all deps completed)
deps = task.get("depends_on", [])
is_actionable = all(task_status.get(dep) == "completed" for dep in deps)
future_tasks.append((phase, task, is_actionable))
if not future_tasks:
if as_json:
print("[]")
else:
print("No upcoming tasks found!")
return
# Sort: in_progress first, then actionable pending, then blocked/waiting
def sort_key(item: tuple[dict, dict, bool]) -> tuple[int, int]:
_, task, is_actionable = item
status_order = {"in_progress": 0, "pending": 1, "blocked": 2}
return (status_order.get(task["status"], 3), 0 if is_actionable else 1)
future_tasks.sort(key=sort_key)
if as_json:
output = [
{
"id": task["id"],
"title": task["title"],
"status": task["status"],
"phase_id": phase["id"],
"phase_name": phase["name"],
"agent_type": task.get("agent_type"),
"skill": task.get("skill"),
"actionable": is_actionable,
"depends_on": task.get("depends_on", []),
}
for phase, task, is_actionable in future_tasks[:count]
]
print(json.dumps(output, indent=2))
return
print(f"\n{bold('Upcoming Tasks:')}\n")
for phase, task, is_actionable in future_tasks[:count]:
task_id = task["id"]
task_title = task["title"]
phase_name = phase["name"]
status = task["status"]
if status == "in_progress":
icon = "🔄"
elif status == "blocked":
icon = "🚫"
elif is_actionable:
icon = "👉"
else:
icon = "⏳"
print(f" {icon} [{task_id}] {task_title}")
status_info = status if status != "pending" else ("ready" if is_actionable else "waiting")
print(f" {dim(f'{phase_name} · {status_info}')}")
print()
[docs]
def cmd_summary(plan: dict, *, as_json: bool = False) -> None:
"""Display summary of plan progress."""
meta = plan.get("meta", {})
summary = plan.get("summary", {})
if as_json:
output = {
"project": meta.get("project"),
"version": meta.get("version"),
**summary,
}
print(json.dumps(output))
return
# Pretty output (default)
project = meta.get("project", "Unknown Project")
version = meta.get("version", "0.0.0")
total = summary.get("total_tasks", 0)
completed = summary.get("completed_tasks", 0)
pct = summary.get("overall_progress", 0)
print(f"\n{bold(f'📋 {project} v{version}')}")
print(f"Overall Progress: {pct:.1f}% ({completed}/{total} tasks)\n")
# Phase-by-phase breakdown
print(bold("Phase Breakdown:"))
for phase in plan.get("phases", []):
progress = phase.get("progress", {})
phase_pct = progress.get("percentage", 0)
phase_completed = progress.get("completed", 0)
phase_total = progress.get("total", 0)
icon = ICONS.get(phase["status"], "❓")
phase_id = phase["id"]
phase_name = phase["name"]
print(f" {icon} Phase {phase_id}: {phase_name}")
print(f" {phase_pct:.1f}% ({phase_completed}/{phase_total} tasks)")
print()
[docs]
def cmd_validate(plan: dict, path: Path, *, as_json: bool = False) -> None:
"""Validate plan against JSON schema."""
schema = load_schema()
try:
jsonschema.validate(plan, schema)
if as_json:
print(json.dumps({"valid": True, "path": str(path)}))
else:
print(f"✅ {path} is valid")
except jsonschema.ValidationError as e:
if as_json:
json_path = ".".join(str(p) for p in e.absolute_path) if e.absolute_path else None
print(json.dumps({"valid": False, "path": str(path), "error": e.message, "json_path": json_path}))
else:
print(f"❌ Validation failed for {path}:")
print(f" {e.message}")
if e.absolute_path:
json_path = ".".join(str(p) for p in e.absolute_path)
print(f" Path: {json_path}")
sys.exit(1)
def _display_special_phase(plan: dict, phase_id: str, phase_name: str, *, as_json: bool = False) -> None:
"""Display a special phase (bugs or deferred)."""
phase = find_phase(plan, phase_id)
if phase is None:
if as_json:
print("null")
else:
print(f"No {phase_name.lower()} phase found!")
return
if as_json:
print(json.dumps(phase, indent=2))
return
tasks = phase.get("tasks", [])
progress = phase.get("progress", {})
completed = progress.get("completed", 0)
total = progress.get("total", 0)
print(f"\n{bold_cyan(f'{phase_name} ({total} tasks)')}")
print(f" {phase['description']}")
if total > 0:
print(f" Completed: {completed}/{total}\n")
else:
print()
if not tasks:
print(f" No {phase_name.lower()} tasks.\n")
return
for task in tasks:
icon = ICONS.get(task["status"], "❓")
task_id = task["id"]
task_title = task["title"]
agent = task.get("agent_type") or "general"
agent_str = f"({agent})" if task.get("agent_type") else ""
skill = task.get("skill")
skill_str = f" [skill: {skill}]" if skill else ""
print(f" {icon} [{task_id}] {task_title} {dim(agent_str)}{dim(skill_str)}")
# Display defer reason if this is a deferred phase and reason exists
if phase_id == "deferred":
defer_reason = task.get("tracking", {}).get("defer_reason")
if defer_reason:
print(f" {dim(f'Reason: {defer_reason}')}")
print()
[docs]
def cmd_bugs(plan: dict, *, as_json: bool = False) -> None:
"""Display bugs phase with all tasks."""
_display_special_phase(plan, "bugs", "Bugs", as_json=as_json)
[docs]
def cmd_deferred(plan: dict, *, as_json: bool = False) -> None:
"""Display deferred phase with all tasks."""
_display_special_phase(plan, "deferred", "Deferred", as_json=as_json)
[docs]
def cmd_ideas(plan: dict, *, as_json: bool = False) -> None:
"""Display ideas phase with all tasks."""
_display_special_phase(plan, "ideas", "Ideas", as_json=as_json)
[docs]
def cmd_table(plan: dict, phase_id: str | None = None, *, as_json: bool = False) -> None:
"""Display plan or phase tasks in table format."""
# Collect tasks to display
tasks_data: list[tuple[dict, dict]] = [] # (phase, task)
if phase_id:
phase = find_phase(plan, phase_id)
if phase is None:
print(f"Phase '{phase_id}' not found!")
return
tasks_data.extend((phase, task) for task in phase.get("tasks", []))
else:
for phase in plan.get("phases", []):
tasks_data.extend((phase, task) for task in phase.get("tasks", []))
if not tasks_data:
if as_json:
print("[]")
else:
print("No tasks found!")
return
if as_json:
output = [
{
"id": task["id"],
"title": task["title"],
"status": task["status"],
"phase": phase["name"],
"agent": task.get("agent_type"),
"skill": task.get("skill"),
}
for phase, task in tasks_data
]
print(json.dumps(output, indent=2))
return
# Calculate column widths
id_width = max(len(task["id"]) for _, task in tasks_data)
title_width = min(40, max(len(task["title"]) for _, task in tasks_data))
status_width = max(len(task["status"]) for _, task in tasks_data)
phase_width = min(15, max(len(phase["name"]) for phase, _ in tasks_data))
agent_width = max(len(task.get("agent_type") or "-") for _, task in tasks_data)
agent_width = min(20, max(5, agent_width))
skill_width = max(len(task.get("skill") or "-") for _, task in tasks_data)
skill_width = min(15, max(5, skill_width))
# Print header
header = (
f"{'ID':<{id_width}} "
f"{'Title':<{title_width}} "
f"{'Status':<{status_width}} "
f"{'Phase':<{phase_width}} "
f"{'Agent':<{agent_width}} "
f"{'Skill':<{skill_width}}"
)
print(f"\n{bold(header)}")
print("─" * len(header))
# Print rows
for phase, task in tasks_data:
icon = ICONS.get(task["status"], "❓")
title = task["title"]
if len(title) > title_width:
title = title[: title_width - 2] + ".."
phase_name = phase["name"]
if len(phase_name) > phase_width:
phase_name = phase_name[: phase_width - 2] + ".."
agent = task.get("agent_type") or "-"
if len(agent) > agent_width:
agent = agent[: agent_width - 2] + ".."
skill = task.get("skill") or "-"
if len(skill) > skill_width:
skill = skill[: skill_width - 2] + ".."
row = (
f"{task['id']:<{id_width}} "
f"{title:<{title_width}} "
f"{icon} {task['status']:<{status_width - 2}} "
f"{phase_name:<{phase_width}} "
f"{agent:<{agent_width}} "
f"{skill:<{skill_width}}"
)
print(row)
print()
print(f"{dim(f'Total: {len(tasks_data)} tasks')}")