"""Development Effort Calculator (WP-0010-T02). Implements Candidate A (labor-cost-anchored), accepted 2026-07-30 (`workplans/TREV-WP-0010-development-effort-calculator.md` T01, `specs/DevelopmentEffortCalculatorConcept.md` §3): human interaction time drives `estimated_effort_days`/`daily_rate` directly; AI token cost becomes `approved_direct_costs`; workplan/task-volume and repo-size metrics are used only as a sanity check on the human-time estimate, never their own dollar figure. This module produces `target_basis` values (`specs/PhaseManifestSpecification.md`) feeding the framework's existing Initial Target formula — it does not compute an Initial Target itself, and it never chooses a Target Multiple (that remains a human judgment call, concept §4). Every function here is pure or reads only from the local filesystem/git history — no network calls, no dependency on the state hub being reachable, consistent with the offline-first design already established elsewhere in this library (`fold.py`, `validation.py`). **1-day manual-work floor (explicit, not silent):** commit-timestamp clustering is, by the concept document's own framing, a floor estimate — it systematically undercounts real effort (thinking/reading time, work that never produced a commit). Any raw estimate below one day is treated as a signal the measurement itself is incomplete, not as a true "this took under a day" fact. `estimate_target_basis` floors the reported value to `MANUAL_EFFORT_FLOOR_DAYS` in that case and attaches a prominent warning recommending a manual override — the floor is a safe fallback so the Phase Manifest never carries an implausibly small figure, not a replacement for a human's own judgment about the real effort involved. """ from __future__ import annotations import re import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any MANUAL_EFFORT_FLOOR_DAYS = 1.0 DEFAULT_SESSION_GAP_HOURS = 2.0 DEFAULT_HOURS_PER_DAY = 8.0 SINGLE_COMMIT_SESSION_MINUTES = 15.0 # v0 placeholder pricing — flagged for refinement once real usage data # exists; not authoritative, callers may override via `model_pricing`. DEFAULT_MODEL_PRICING_USD_PER_TOKEN: dict[str, dict[str, float]] = { "default": {"input": 3e-6, "output": 15e-6}, } # Paths that inflate the repo-size signal without reflecting real # authored effort (specs/DevelopmentEffortCalculatorConcept.md §2c's # flagged distortion). Matched against any path component. GENERATED_VENDORED_EXCLUDES = frozenset( { "node_modules", ".venv", "venv", "__pycache__", ".git", "dist", "build", ".pytest_cache", "vendor", "site-packages", ".egg-info", ".mypy_cache", ".ruff_cache", } ) @dataclass(frozen=True) class CommitTimeEstimate: """Result of session-gap clustering over a repo's commit timestamps.""" raw_hours: float session_count: int commit_count: int @dataclass(frozen=True) class WorkplanTaskCounts: """Workplan/task volume, used only as a sanity-check signal (§2b).""" finished_workplans: int total_tasks: int finished_tasks: int @dataclass(frozen=True) class RepoSizeMetrics: """File/line counts, used only as a sanity-check signal (§2c).""" file_count: int line_count: int @dataclass(frozen=True) class TokenCost: """AI token cost — the strongest, least speculative input (§2d).""" tokens_in: int tokens_out: int usd: float @dataclass(frozen=True) class TargetBasisEstimate: """Proposed `phase.target_basis` values, with visible derivation. `warnings` and `derivation` exist so the output is never a black-box dollar figure — per `specs/TargetRevenueLicenseConcept.md` §4.6's "transparent, non-gameable" goal, showing which inputs produced which numbers is part of this calculator's job, not an optional extra. """ estimated_effort_days: float daily_rate: float approved_direct_costs: float warnings: list[str] = field(default_factory=list) derivation: dict[str, Any] = field(default_factory=dict) def cluster_commit_hours( repo_path: str | Path, session_gap_hours: float = DEFAULT_SESSION_GAP_HOURS, since: str | None = None, until: str | None = None, ) -> CommitTimeEstimate: """Group commit timestamps into sessions (gap-based clustering). A session is a run of commits with no gap larger than `session_gap_hours` between consecutive commits. Each session contributes its own wall-clock span; a session of a single commit (span zero) contributes a fixed `SINGLE_COMMIT_SESSION_MINUTES` floor, since a lone commit still took some real, non-zero time to produce. This is an explicit floor estimate (concept §2a), not a true count — time spent reading, thinking, or working without committing is invisible to it by construction. `since`/`until` (any `git log --since`/`--until`-accepted date string, e.g. `"2026-03-01"`) scope the estimate to a date range — required whenever the candidate Milestone Release is one bounded workplan within a repo whose overall history spans much more unrelated work than that one Phase. Whole-repo history (the default, no scoping) is only correct when the repo's entire history *is* the candidate. """ cmd = ["git", "-C", str(repo_path), "log", "--all", "--format=%at"] if since: cmd += ["--since", since] if until: cmd += ["--until", until] result = subprocess.run(cmd, capture_output=True, text=True, check=True) timestamps = sorted(int(line) for line in result.stdout.splitlines() if line.strip()) if not timestamps: return CommitTimeEstimate(raw_hours=0.0, session_count=0, commit_count=0) gap_seconds = session_gap_hours * 3600 sessions: list[tuple[int, int]] = [] session_start = timestamps[0] session_end = timestamps[0] for ts in timestamps[1:]: if ts - session_end > gap_seconds: sessions.append((session_start, session_end)) session_start = ts session_end = ts sessions.append((session_start, session_end)) total_hours = 0.0 for start, end in sessions: span_hours = (end - start) / 3600 if span_hours <= 0: span_hours = SINGLE_COMMIT_SESSION_MINUTES / 60 total_hours += span_hours return CommitTimeEstimate( raw_hours=total_hours, session_count=len(sessions), commit_count=len(timestamps) ) _TASK_BLOCK_RE = re.compile(r"```task\n(.*?)\n```", re.DOTALL) _FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---", re.DOTALL) _STATUS_FIELD_RE = re.compile(r'^status:\s*"?([^"\n]+)"?\s*$', re.MULTILINE) _UPDATED_FIELD_RE = re.compile(r'^updated:\s*"?([0-9]{4}-[0-9]{2}-[0-9]{2})"?\s*$', re.MULTILINE) _CREATED_FIELD_RE = re.compile(r'^created:\s*"?([0-9]{4}-[0-9]{2}-[0-9]{2})"?\s*$', re.MULTILINE) def workplan_task_counts( repo_path: str | Path, since: str | None = None, until: str | None = None ) -> WorkplanTaskCounts: """Count finished workplans and tasks under `/workplans/`. Follows this repo's own workplan convention (YAML-ish frontmatter with a `status:` field, ```task fenced blocks each with their own `status:` line) rather than requiring state-hub connectivity — works on any repo using this convention, degrades to zero (not an error) for repos with no `workplans/` directory at all. Scans recursively (not just the top level), since several repos in this ecosystem archive finished workplans under `workplans/archived/` rather than deleting them — a non-recursive scan would silently undercount exactly the finished work this function exists to measure. `since`/`until` (`YYYY-MM-DD` strings) restrict counting to workplans whose frontmatter `updated` (falling back to `created`) date falls within the window — a workplan file with neither field is always counted, since there is nothing to filter it by. Pass matching `since`/`until` values to `cluster_commit_hours` when scoping a candidate to one bounded workplan within a repo whose overall history spans much more; otherwise a whole-repo workplan count compared against a date-scoped time estimate produces a misleading sanity-check signal (falsely flagging under-counting). """ workplans_dir = Path(repo_path) / "workplans" if not workplans_dir.is_dir(): return WorkplanTaskCounts(finished_workplans=0, total_tasks=0, finished_tasks=0) finished_workplans = 0 total_tasks = 0 finished_tasks = 0 for md_file in sorted(workplans_dir.rglob("*.md")): try: text = md_file.read_text(encoding="utf-8", errors="ignore") except OSError: continue frontmatter_match = _FRONTMATTER_RE.match(text) frontmatter = frontmatter_match.group(1) if frontmatter_match else "" if since or until: date_match = _UPDATED_FIELD_RE.search(frontmatter) or _CREATED_FIELD_RE.search(frontmatter) if date_match: date_str = date_match.group(1) if since and date_str < since: continue if until and date_str > until: continue if frontmatter_match: status_match = _STATUS_FIELD_RE.search(frontmatter) if status_match and status_match.group(1).strip() == "finished": finished_workplans += 1 for block in _TASK_BLOCK_RE.findall(text): total_tasks += 1 task_status_match = _STATUS_FIELD_RE.search(block) if task_status_match and task_status_match.group(1).strip() == "done": finished_tasks += 1 return WorkplanTaskCounts( finished_workplans=finished_workplans, total_tasks=total_tasks, finished_tasks=finished_tasks, ) def repo_size_metrics( repo_path: str | Path, extra_excludes: frozenset[str] = frozenset() ) -> RepoSizeMetrics: """File/line counts, excluding generated/vendored/dependency paths. Exclusion is by path-component membership (any directory named `node_modules`, `.venv`, etc. anywhere in a file's path is skipped entirely), not just top-level — this catches nested vendored trees, not only ones at the repo root. """ excludes = GENERATED_VENDORED_EXCLUDES | extra_excludes file_count = 0 line_count = 0 for path in Path(repo_path).rglob("*"): if not path.is_file(): continue if any(part in excludes for part in path.parts): continue file_count += 1 try: with open(path, "rb") as f: line_count += sum(1 for _ in f) except OSError: continue return RepoSizeMetrics(file_count=file_count, line_count=line_count) def token_cost_usd( tokens_in: int, tokens_out: int, model: str = "default", model_pricing: dict[str, dict[str, float]] | None = None, ) -> float: """AI token cost in USD for the given token counts and model. `model_pricing` defaults to `DEFAULT_MODEL_PRICING_USD_PER_TOKEN` — a v0 placeholder table, not authoritative pricing; callers with real per-model rates should pass their own table. Unknown models fall back to `"default"` rather than raising, since the exact model used is a `get_token_summary` reporting detail this module has no way to validate against a live price list. """ pricing_table = model_pricing or DEFAULT_MODEL_PRICING_USD_PER_TOKEN pricing = pricing_table.get(model, pricing_table["default"]) return tokens_in * pricing["input"] + tokens_out * pricing["output"] def estimate_target_basis( *, commit_time: CommitTimeEstimate, workplan_tasks: WorkplanTaskCounts, repo_size: RepoSizeMetrics, token_cost: TokenCost, daily_rate: float, hours_per_day: float = DEFAULT_HOURS_PER_DAY, ) -> TargetBasisEstimate: """Combine the four metric families into proposed `target_basis` values. Candidate A (labor-cost-anchored): `estimated_effort_days` comes directly from `commit_time`; `approved_direct_costs` comes directly from `token_cost`; `workplan_tasks`/`repo_size` never produce their own dollar figure — they only trigger sanity-check warnings when they suggest the time estimate is implausibly low for the amount of finished, real work on record. """ warnings: list[str] = [] derivation: dict[str, Any] = {} raw_effort_days = commit_time.raw_hours / hours_per_day derivation["raw_commit_clustered_hours"] = round(commit_time.raw_hours, 3) derivation["raw_commit_clustered_days"] = round(raw_effort_days, 3) derivation["sessions"] = commit_time.session_count derivation["commits"] = commit_time.commit_count complexity_units_workplans = ( workplan_tasks.finished_workplans * 3 + workplan_tasks.finished_tasks * 1 ) derivation["finished_workplans"] = workplan_tasks.finished_workplans derivation["finished_tasks"] = workplan_tasks.finished_tasks derivation["complexity_units_workplans"] = complexity_units_workplans derivation["file_count"] = repo_size.file_count derivation["line_count"] = repo_size.line_count effort_days = raw_effort_days if effort_days < MANUAL_EFFORT_FLOOR_DAYS: warnings.append( f"Raw commit-clustered estimate was {raw_effort_days:.3f} day(s), below the " f"{MANUAL_EFFORT_FLOOR_DAYS:.1f}-day floor. Floored to " f"{MANUAL_EFFORT_FLOOR_DAYS:.1f} day(s) rather than reporting an implausibly " "small figure. This is very likely a measurement gap — commit-timestamp " "clustering is a floor estimate by design " "(specs/DevelopmentEffortCalculatorConcept.md §2a), not a true reflection of " "effort — and should usually be compensated for by a manual override of " "estimated_effort_days rather than trusting this floored value at face value." ) effort_days = MANUAL_EFFORT_FLOOR_DAYS # Sanity-check signal (§2b/§2c): a real, non-trivial finished-work # history alongside a still-low time estimate is itself evidence the # time input is under-counting, independent of the floor above. if complexity_units_workplans >= 15 and raw_effort_days < MANUAL_EFFORT_FLOOR_DAYS * 3: warnings.append( f"{workplan_tasks.finished_workplans} finished workplan(s) and " f"{workplan_tasks.finished_tasks} finished task(s) on record suggest more real " f"effort than the {raw_effort_days:.2f}-day raw time estimate reflects — treat " "the time estimate as under-counted and consider a manual override." ) derivation["approved_direct_costs_usd"] = round(token_cost.usd, 2) derivation["tokens_in"] = token_cost.tokens_in derivation["tokens_out"] = token_cost.tokens_out return TargetBasisEstimate( estimated_effort_days=round(effort_days, 3), daily_rate=daily_rate, approved_direct_costs=round(token_cost.usd, 2), warnings=warnings, derivation=derivation, ) def calculate_target_basis( repo_path: str | Path, daily_rate: float, tokens_in: int = 0, tokens_out: int = 0, model: str = "default", session_gap_hours: float = DEFAULT_SESSION_GAP_HOURS, hours_per_day: float = DEFAULT_HOURS_PER_DAY, extra_excludes: frozenset[str] = frozenset(), model_pricing: dict[str, dict[str, float]] | None = None, since: str | None = None, until: str | None = None, ) -> TargetBasisEstimate: """End-to-end convenience entry point: repo path + token counts in, a proposed `target_basis` estimate (with warnings and derivation) out. `tokens_in`/`tokens_out` are caller-supplied (e.g. from the state hub's `get_token_summary`) rather than fetched here, since this module has no MCP/network client of its own — keeping it offline-first like the rest of this library. `since`/`until` scope both commit-time clustering and workplan/task counting to the same date window — pass both whenever the candidate Milestone Release is one bounded workplan within a repo whose overall history spans much more (see `cluster_commit_hours`/ `workplan_task_counts` docstrings); leave unset only when the repo's entire history genuinely is the candidate. Repo size (file/line counts) is not date-scoped — it reflects the repo's current state regardless of when each file was last touched, which is intentional (`target_basis` describes the Milestone Release's actual delivered size, not a historical snapshot) but worth noting as an asymmetry. """ commit_time = cluster_commit_hours( repo_path, session_gap_hours=session_gap_hours, since=since, until=until ) workplan_tasks = workplan_task_counts(repo_path, since=since, until=until) repo_size = repo_size_metrics(repo_path, extra_excludes=extra_excludes) usd = token_cost_usd(tokens_in, tokens_out, model=model, model_pricing=model_pricing) token_cost = TokenCost(tokens_in=tokens_in, tokens_out=tokens_out, usd=usd) return estimate_target_basis( commit_time=commit_time, workplan_tasks=workplan_tasks, repo_size=repo_size, token_cost=token_cost, daily_rate=daily_rate, hours_per_day=hours_per_day, )