Files
state-hub/api/schemas/task.py
tegwick af3fdfde80 feat(token-tracking): introduce token note taxonomy (measured/userbased/workplan/heuristic)
Tier 1 (exact counts) now defaults to note="measured" instead of null,
signalling the counts were read from the Claude Code status bar.
Callers can pass note="userbased" when a human provided the numbers.

  measured  — agent read exact counts from the Claude Code status bar
  userbased — counts provided by a human
  workplan  — prorated from workplan total across task count
  heuristic — server fallback, 1000/500, no agent input

Added token_note field to TaskUpdate schema and exposed note param on
update_task_status and record_interactive_task MCP tools.
TOOLS.md documents the full taxonomy. 185 tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-29 18:47:40 +02:00

84 lines
2.9 KiB
Python

import uuid
from datetime import date, datetime
from typing import Self
from pydantic import BaseModel, ConfigDict, model_validator
from api.models.task import TaskPriority, TaskStatus
class TaskCreate(BaseModel):
workstream_id: uuid.UUID
title: str
description: str | None = None
status: TaskStatus = TaskStatus.todo
priority: TaskPriority = TaskPriority.medium
assignee: str | None = None
due_date: date | None = None
blocking_reason: str | None = None
needs_human: bool = False
intervention_note: str | None = None
parent_task_id: uuid.UUID | None = None
@model_validator(mode="after")
def intervention_note_required_when_flagged(self) -> Self:
if self.needs_human and not self.intervention_note:
raise ValueError("intervention_note is required when needs_human is True")
return self
class TaskUpdate(BaseModel):
title: str | None = None
description: str | None = None
status: TaskStatus | None = None
priority: TaskPriority | None = None
assignee: str | None = None
due_date: date | None = None
blocking_reason: str | None = None
needs_human: bool | None = None
intervention_note: str | None = None
parent_task_id: uuid.UUID | None = None
# Token passthrough — three tiers (highest precision wins):
# 1. tokens_in + tokens_out → exact counts; note defaults to "measured"
# 2. workplan_tokens_in + workplan_tokens_out → prorated across task count (note="workplan")
# 3. neither provided, status=done → heuristic 1000/500 (note="heuristic")
# token_note overrides the auto-assigned note for Tier 1 only (e.g. "userbased")
tokens_in: int | None = None
tokens_out: int | None = None
workplan_tokens_in: int | None = None
workplan_tokens_out: int | None = None
token_note: str | None = None
model: str | None = None
agent: str | None = None
session_id: str | None = None
@model_validator(mode="after")
def blocking_reason_required_when_blocked(self) -> Self:
if self.status == TaskStatus.blocked and not self.blocking_reason:
raise ValueError("blocking_reason is required when status is blocked")
return self
@model_validator(mode="after")
def intervention_note_required_when_flagged(self) -> Self:
if self.needs_human and not self.intervention_note:
raise ValueError("intervention_note is required when needs_human is True")
return self
class TaskRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
workstream_id: uuid.UUID
title: str
description: str | None = None
status: TaskStatus
priority: TaskPriority
assignee: str | None = None
due_date: date | None = None
blocking_reason: str | None = None
needs_human: bool
intervention_note: str | None = None
parent_task_id: uuid.UUID | None = None
created_at: datetime
updated_at: datetime