Skip to content

Task Tracker

task_tracker — Built-in task tracking with progress, cost, scheduling, dependencies, and HITL review.

Cluster: Uncategorised | Type: component | MCP Tools: 30

Overview

Built-in task tracking with progress monitoring, cost accumulation, scheduling, dependency management (with cycle detection), and human-in-the-loop (HITL) review workflows. Uses SQLite persistence and enforces a strict state machine for task lifecycle transitions.

The registry-facing block and MCP layer expose 30 operations covering CRUD, lifecycle, scheduling, dependencies, HITL review, analytics, grounding introspection, and read-only source health.

Pilot/local scope

task_tracker is designed for local MCP installs, solo-founder workflows, demos, and design-partner pilots. Its default persistence is a local SQLite database (TASK_TRACKER_DB_PATH, otherwise ~/.task_tracker/tasks.db). This is useful for single-machine task state, audit trails, dependency tracking, and review queues, but it is not a shared multi-user SaaS task-management backend. For paid team deployments, put it behind a single writer or migrate the store to a server database with explicit tenant isolation, migrations, backups, retention policy, and operational monitoring.

When to use:

  • Tracking work items with lifecycle state enforcement (pending, claimed, completed, failed)
  • Managing task dependencies with automatic cycle detection to prevent deadlocks
  • Accumulating cost and progress metrics for reporting and budget control

Example:

from mvp.task_tracker import TaskTrackerBlock, TaskTrackerInput

block = TaskTrackerBlock()
result = block.infer(TaskTrackerInput(
    op="create_task",
    title="Fix goal decomposition depth",
    priority=2,  # int (higher = more urgent)
    tags="goal_engine,bug",  # comma-separated string
))
# result.value -> TaskTrackerOutput with task.task_id, task.status="pending"

Caveats and known limitations:

  • SQLite persistence is appropriate for local pilots, not high-concurrency multi-host coordination

  • ResourceBounds guard is optional — operations run unbounded if no limits are configured

  • State machine transitions are enforced by the underlying store, not documented in the block interface
  • Dependency cycles are blocked on insertion; deleting a dependency target removes its dependency edges
  • Incomplete dependencies block task claiming, but cross-process scheduling still depends on the configured RQ/worker setup
  • When adapt_rq is unavailable, scheduling honestly degrades to SQLite-only timestamp storage with degradation_reason="adapt_rq_unavailable_local_timestamp_only"
  • source_health reports sanitized SQLite, completion-gate, scheduler, and capability state without mutating the database or calling an LLM
  • complete_task responses expose completion_state using only verified, qualified-draft, or blocked-escalated
  • _dispatch() returns raw dicts from the store — _build_output() infers output shape, which can be fragile if store schema changes

Works well with: work_loop, token_budget, email_colleague

Public API

TaskStatus(str, Enum)

ReviewStatus(str, Enum)

TaskTrackerInput(BaseModel)

Tier 1 input for basic task tracking operations.

Field Type Default
op TaskTrackerOp required
task_id str ''
title str ''
description str ''
status str ''
priority int 0
progress_pct float 0.0
cost_usd float 0.0
cost_compute_seconds float 0.0
assigned_agent str ''
component str ''
tags str ''
requires_review bool False
review_comment str ''
reviewer str ''
parent_id str \| None None
scheduled_at str ''
depends_on_id str ''
dep_type str 'blocks'
blocker_reason str ''
filter_status str ''
filter_component str ''
filter_tags str ''
filter_review_status str ''
limit int 100
offset int 0
error_message str ''
output_format str 'json'
metadata_json str '{}'
confirm bool False
project_id str ''
filter_project str ''
params dict[str, Any] Field(default_factory=dict)

TaskTrackerOutput(BaseModel)

Tier 1 output for task tracking operations.

Field Type Default
op str ''
task dict[str, Any] Field(default_factory=dict)
tasks list[dict[str, Any]] Field(default_factory=list)
dependencies list[dict[str, Any]] Field(default_factory=list)
graph dict[str, Any] Field(default_factory=dict)
reviews list[dict[str, Any]] Field(default_factory=list)
stats dict[str, Any] Field(default_factory=dict)
cost_summary dict[str, Any] Field(default_factory=dict)
estimated_completion str ''
ok bool True
created bool False
updated bool False
deleted bool False
scheduled bool False
error str ''
count int 0
message str ''
metadata dict[str, Any] Field(default_factory=dict)
completion_state str 'qualified-draft'
degraded bool False
degradation_reason str \| None None
warning_card dict[str, Any] \| None None
requires_review bool False
agentic_evidence dict[str, Any] \| None None

TaskTrackerBlock(AIBlock['TaskTrackerInput', 'TaskTrackerOutput', dict])

Registry-facing task tracker with SQLite persistence.

Field Type Default
name str 'task_tracker'
state dict field(default_factory=dict)
db_path str field(default_factory=lambda: os.environ.get('TASK_TRACKER_DB_PATH', _DEFAULT_DB))
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
planner Any field(default=None)

Methods:

infer(data: TaskTrackerInput) -> Result[TaskTrackerOutput]

MCP Tools

Operation Source
create_task task_tracker_mcp
get_task task_tracker_mcp
list_tasks task_tracker_mcp
update_task task_tracker_mcp
delete_task task_tracker_mcp
claim_task task_tracker_mcp
complete_task task_tracker_mcp
fail_task task_tracker_mcp
block_task task_tracker_mcp
unblock_task task_tracker_mcp
update_progress task_tracker_mcp
schedule_task task_tracker_mcp
cancel_scheduled task_tracker_mcp
list_scheduled task_tracker_mcp
check_due_tasks task_tracker_mcp
add_dependency task_tracker_mcp
remove_dependency task_tracker_mcp
get_dependencies task_tracker_mcp
dependency_graph task_tracker_mcp
request_review task_tracker_mcp
approve_task task_tracker_mcp
reject_task task_tracker_mcp
list_pending_reviews task_tracker_mcp
review_history task_tracker_mcp
task_stats task_tracker_mcp
estimate_completion task_tracker_mcp
cost_summary task_tracker_mcp
info task_tracker_mcp
list_patterns task_tracker_mcp
source_health task_tracker_mcp