feat: add issue_usage tracking and metrics for AI ticket enhancement and planning

Add `issue_usage` table with columns for user_uid, tokens, cost, and latency metrics, including a unique index on user_uid. Wire `accumulate_usage` into `enhance_ticket` and `generate_plan` to capture per-request AI usage, persist totals via `add_issue_usage` in both `IssueCreateService` and `PlanningReportService`, and expose aggregated usage as metric cards through `IssueTrackerService.collect_metrics`. Update planning API docs summary to reflect the new phased implementation document format.
This commit is contained in:
2026-06-19 11:59:40 +00:00
parent 7bbaf51450
commit ff3cbbbfe2
16 changed files with 503 additions and 122 deletions
+34
View File
@@ -251,6 +251,17 @@ def get_news_usage() -> dict:
return _get_usage("news_usage", NEWS_USAGE_KEY)
ISSUE_USAGE_KEY = "issues"
def add_issue_usage(totals: dict) -> None:
_add_usage("issue_usage", ISSUE_USAGE_KEY, totals)
def get_issue_usage() -> dict:
return _get_usage("issue_usage", ISSUE_USAGE_KEY)
def record_activity(user_uid: str, action: str) -> int:
if not user_uid or not action or "user_activity" not in db.tables:
return 0
@@ -941,6 +952,29 @@ def init_db():
except Exception as e:
logger.warning(f"Could not create unique index on news_usage: {e}")
issue_usage = get_table("issue_usage")
for column, example in (
("user_uid", ""),
("calls", 0),
("prompt_tokens", 0),
("completion_tokens", 0),
("total_tokens", 0),
("cost_usd", 0.0),
("upstream_latency_ms", 0.0),
("total_latency_ms", 0.0),
("updated_at", ""),
):
if not issue_usage.has_column(column):
issue_usage.create_column_by_example(column, example)
try:
if "issue_usage" in db.tables:
db.query(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_issue_usage_user "
"ON issue_usage (user_uid)"
)
except Exception as e:
logger.warning(f"Could not create unique index on issue_usage: {e}")
email_accounts = get_table("email_accounts")
for column, example in (
("uid", ""),
+2 -2
View File
@@ -3976,7 +3976,7 @@ four ways to sign requests.
method="POST",
path="/issues/planning",
title="Generate a tickets planning report",
summary="Enqueue a grouped, ordered markdown planning report of all open tickets. Admin only.",
summary="Enqueue a complete, phased markdown implementation document covering every open ticket, with each ticket's full description reproduced verbatim plus implementation steps and acceptance criteria. Admin only.",
auth="admin",
encoding="form",
ajax=True,
@@ -4001,7 +4001,7 @@ four ways to sign requests.
"kind": "planning",
"status": "done",
"download_url": "/issues/planning/PLANNING_JOB_UID/download",
"markdown": "# Open Tickets Planning\n\n...",
"markdown": "# Open Tickets Implementation Plan\n\n...",
"ai_used": True,
"issue_count": 12,
"bytes_out": 4096,
+3 -1
View File
@@ -10,6 +10,7 @@ import httpx
from devplacepy import stealth
from devplacepy.config import INTERNAL_GATEWAY_URL
from devplacepy.services.gitea.config import HTTP_TIMEOUT_SECONDS, GiteaConfig
from devplacepy.services.openai_gateway.usage import accumulate_usage
logger = logging.getLogger(__name__)
@@ -68,7 +69,7 @@ def _parse(text: str) -> tuple[str, str] | None:
async def enhance_ticket(
title: str, description: str, config: GiteaConfig
title: str, description: str, config: GiteaConfig, totals: dict | None = None
) -> EnhancedTicket:
title = title.strip()
description = description.strip()
@@ -97,6 +98,7 @@ async def enhance_ticket(
INTERNAL_GATEWAY_URL, json=payload, headers=headers
)
response.raise_for_status()
accumulate_usage(totals, response)
content = (
response.json()
.get("choices", [{}])[0]
+94 -44
View File
@@ -6,39 +6,69 @@ import httpx
from devplacepy import stealth
from devplacepy.config import INTERNAL_GATEWAY_URL
from devplacepy.services.gitea.config import HTTP_TIMEOUT_SECONDS, GiteaConfig
from devplacepy.services.gitea.config import GiteaConfig
from devplacepy.services.openai_gateway.usage import accumulate_usage
logger = logging.getLogger(__name__)
MAX_TOKENS = 3000
MAX_TOKENS = 32000
MAX_ISSUES = 50
BODY_EXCERPT_MAX = 600
PLAN_MAX = 80000
BODY_MAX = 40000
PLAN_MAX = 600000
PLANNING_TIMEOUT_SECONDS = 600.0
UNGROUPED_LABEL = "General"
SYSTEM_PROMPT = (
"You are a senior engineering lead for DevPlace, a social network for software "
"developers. You are given the full list of OPEN issue tickets from a Gitea tracker. "
"Produce a single, clear delivery plan in GitHub-flavored markdown that GROUPS the "
"tickets into logical workstreams and ORDERS them so dependencies and high-impact work "
"come first. Never invent tickets, numbers, or facts that are not in the input.\n\n"
"Return ONLY markdown, using EXACTLY this structure:\n"
"# Open Tickets Planning\n"
"A short paragraph summarising the overall plan and how many tickets it covers.\n\n"
"Then one level-2 heading per group (a workstream or theme). Under each group, an "
"ordered (numbered) list of its tickets. Each list item MUST start with the ticket "
"number as '#N', then the title, then a single sentence explaining the chosen order "
"or the work to do, for example:\n"
"1. #12 Fix the login redirect - blocks every authenticated flow, do first.\n\n"
"End with a level-2 heading '## Suggested Order' giving a flat numbered list of every "
"ticket number in the recommended execution order."
"You are a senior engineering lead for DevPlace, a server-rendered social network for "
"software developers (FastAPI + Jinja2 + pure ES6 modules, SQLite via the dataset "
"library). You are given the COMPLETE, VERBATIM text of every OPEN issue ticket from a "
"Gitea tracker. Your job is to produce ONE single, exhaustive implementation document "
"that an autonomous coding agent (Claude) can execute end to end in one shot, ticket by "
"ticket, with NO further questions and NO missing information.\n\n"
"ABSOLUTE RULES:\n"
"- Never invent tickets, numbers, facts, requirements, or constraints that are not in "
"the input. If a detail is missing from a ticket, say so explicitly rather than guessing.\n"
"- PRESERVE EVERY DETAIL from each ticket description. Do not summarise away, shorten, or "
"drop any requirement, reproduction step, acceptance criterion, edge case, link, code "
"snippet, or example. Reproduce the full original requirement text for each ticket "
"verbatim in its 'Original ticket' block, then expand on it - never replace it.\n"
"- The document must be self-contained and directly executable: enough detail that the "
"agent never has to ask a clarifying question.\n"
"- Output ONLY GitHub-flavored markdown. No preamble, no commentary outside the document.\n\n"
"Use EXACTLY this structure:\n\n"
"# Open Tickets Implementation Plan\n"
"One paragraph stating how many tickets are covered, the phasing strategy, and the "
"ordering rationale (dependencies and high-impact/blocking work first).\n\n"
"## Execution Order\n"
"A flat numbered list of every ticket number in the exact recommended execution order, "
"each line as '#N - <title>'.\n\n"
"Then group the work into PHASES, one '## Phase K: <name>' heading per phase, ordered so "
"blocking and high-impact work comes first. Start each phase with a one-paragraph goal and "
"any prerequisites (which earlier phases or tickets must land first).\n\n"
"Inside each phase, for EVERY ticket it contains, emit a '### #N <title>' subsection with "
"ALL of the following labelled blocks, in this order:\n"
"- **Original ticket**: the complete, unedited ticket description text, quoted verbatim "
"(use a blockquote). Include labels.\n"
"- **Goal**: what 'done' means for this ticket in one or two sentences.\n"
"- **Dependencies**: other ticket numbers or phases this depends on, or 'None'.\n"
"- **Affected areas / files**: the concrete modules, routers, templates, JS, CSS, services, "
"schemas, and docs likely to change (infer from the DevPlace architecture; name real-looking "
"paths where the ticket implies them, and mark anything uncertain as 'to confirm').\n"
"- **Implementation steps**: a numbered, ordered, concrete checklist the agent follows to "
"implement the ticket across the full stack (data layer, server, view, agent/Devii, docs, "
"SEO as applicable). Be specific and actionable, not generic.\n"
"- **Acceptance criteria**: a checklist of verifiable conditions that prove the ticket is "
"complete, derived strictly from the ticket plus the implementation steps.\n"
"- **Risks / open questions**: anything ambiguous in the ticket the agent should watch for; "
"write 'None' if the ticket is fully specified.\n\n"
"Be as long and detailed as necessary. Do not truncate. Completeness beats brevity."
)
def _excerpt(text: str) -> str:
def _body(text: str) -> str:
text = (text or "").strip().replace("\r\n", "\n")
if len(text) > BODY_EXCERPT_MAX:
return text[:BODY_EXCERPT_MAX].rstrip() + "..."
if len(text) > BODY_MAX:
return text[:BODY_MAX].rstrip() + "\n\n[description truncated at safety limit]"
return text
@@ -70,55 +100,74 @@ def _fallback(issues: list[dict]) -> str:
key=lambda name: (name == UNGROUPED_LABEL, name.lower()),
)
lines: list[str] = ["# Open Tickets Planning", ""]
lines: list[str] = ["# Open Tickets Implementation Plan", ""]
lines.append(
f"Deterministic plan covering {len(issues)} open ticket"
f"{'' if len(issues) == 1 else 's'}, grouped by primary label and ordered by "
"ticket number within each group."
"ticket number within each group. The full original description of every ticket is "
"reproduced verbatim below."
)
lines.append("")
flat_order: list[int] = []
for name in ordered_group_names:
lines.append("## Execution Order")
for position, issue in enumerate(issues, start=1):
number = int(issue.get("number", 0))
title = str(issue.get("title", "")).strip() or "Untitled"
lines.append(f"{position}. #{number} - {title}")
lines.append("")
for index, name in enumerate(ordered_group_names, start=1):
members = sorted(groups[name], key=lambda i: int(i.get("number", 0)))
lines.append(f"## {name}")
for position, issue in enumerate(members, start=1):
lines.append(f"## Phase {index}: {name}")
lines.append("")
for issue in members:
number = int(issue.get("number", 0))
title = str(issue.get("title", "")).strip() or "Untitled"
excerpt = _excerpt(issue.get("body", ""))
rationale = excerpt or "No description provided."
lines.append(f"{position}. #{number} {title} - {rationale}")
flat_order.append(number)
lines.append("")
lines.append("## Suggested Order")
for position, number in enumerate(flat_order, start=1):
lines.append(f"{position}. #{number}")
labels = ", ".join(_labels(issue)) or "none"
body = _body(issue.get("body", "")) or "No description provided."
lines.append(f"### #{number} {title}")
lines.append("")
lines.append(f"**Labels**: {labels}")
lines.append("")
lines.append("**Original ticket**:")
lines.append("")
for body_line in body.split("\n"):
lines.append(f"> {body_line}")
lines.append("")
lines.append("")
return "\n".join(lines)[:PLAN_MAX]
def _user_message(issues: list[dict]) -> str:
parts: list[str] = ["Open tickets to plan:", ""]
parts: list[str] = [
f"There are {len(issues)} open tickets. Their complete, verbatim descriptions "
"follow. Produce the full implementation document covering every one of them.",
"",
]
for issue in issues:
number = int(issue.get("number", 0))
title = str(issue.get("title", "")).strip() or "Untitled"
labels = ", ".join(_labels(issue)) or "none"
excerpt = _excerpt(issue.get("body", ""))
parts.append(f"#{number} {title}")
body = _body(issue.get("body", ""))
parts.append(f"===== TICKET #{number} =====")
parts.append(f"title: {title}")
parts.append(f"labels: {labels}")
parts.append(f"description: {excerpt or 'none'}")
parts.append("description (verbatim):")
parts.append(body or "none")
parts.append("")
return "\n".join(parts)
async def generate_plan(
issues: list[dict], config: GiteaConfig
issues: list[dict], config: GiteaConfig, totals: dict | None = None
) -> tuple[str, bool]:
issues = list(issues)[:MAX_ISSUES]
if not issues:
return "# Open Tickets Planning\n\nThere are no open tickets to plan.\n", False
return (
"# Open Tickets Implementation Plan\n\nThere are no open tickets to plan.\n",
False,
)
if not config.ai_enhance:
return _fallback(issues), False
@@ -136,11 +185,12 @@ async def generate_plan(
headers["Authorization"] = f"Bearer {config.ai_key}"
try:
async with stealth.stealth_async_client(timeout=HTTP_TIMEOUT_SECONDS) as client:
async with stealth.stealth_async_client(timeout=PLANNING_TIMEOUT_SECONDS) as client:
response = await client.post(
INTERNAL_GATEWAY_URL, json=payload, headers=headers
)
response.raise_for_status()
accumulate_usage(totals, response)
content = (
response.json()
.get("choices", [{}])[0]
+7 -1
View File
@@ -2,10 +2,12 @@
import logging
from devplacepy.database import get_issue_usage
from devplacepy.services.base import BaseService
from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.client import GiteaError
from devplacepy.services.gitea.config import CONFIG_FIELDS, gitea_config
from devplacepy.services.openai_gateway.usage import usage_metric_cards
from devplacepy.utils import create_notification
logger = logging.getLogger(__name__)
@@ -19,13 +21,17 @@ class IssueTrackerService(BaseService):
description = (
"Holds the Gitea connection and AI settings for the issue tracker and polls every "
"tracked issue for developer replies and status changes, notifying the original "
"reporter when their ticket is updated."
"reporter when their ticket is updated. Reports the aggregated AI usage spent on "
"ticket enhancement and planning."
)
config_fields = CONFIG_FIELDS
def __init__(self):
super().__init__(name="issue_tracker", interval_seconds=120)
def collect_metrics(self) -> dict:
return {"stats": usage_metric_cards(get_issue_usage())}
async def run_once(self) -> None:
config = gitea_config()
if not config.is_configured:
@@ -2,12 +2,13 @@
import logging
from devplacepy.database import get_table
from devplacepy.database import add_issue_usage, get_table
from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.client import GiteaError
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.gitea.enhance import enhance_ticket
from devplacepy.services.jobs.base import JobService
from devplacepy.services.openai_gateway.usage import new_usage_totals
from devplacepy.utils import create_notification
logger = logging.getLogger(__name__)
@@ -50,7 +51,10 @@ class IssueCreateService(JobService):
if not config.is_configured:
raise GiteaError("Gitea integration is not configured", status=503)
enhanced = await enhance_ticket(title, description, config)
usage_totals = new_usage_totals()
enhanced = await enhance_ticket(title, description, config, usage_totals)
if usage_totals["calls"]:
add_issue_usage(usage_totals)
body = _attribution(enhanced.body, user["username"])
issue = await runtime.get_client().create_issue(enhanced.title, body)
+10 -4
View File
@@ -6,11 +6,13 @@ from pathlib import Path
from devplacepy.attachments import _directory_for
from devplacepy.config import PLANNING_REPORTS_DIR
from devplacepy.database import add_issue_usage
from devplacepy.services.gitea import runtime
from devplacepy.services.gitea.client import STATE_OPEN, GiteaError
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.gitea.planning import MAX_ISSUES, generate_plan
from devplacepy.services.jobs.base import JobService
from devplacepy.services.openai_gateway.usage import new_usage_totals
from devplacepy.utils import slugify
logger = logging.getLogger(__name__)
@@ -22,9 +24,10 @@ class PlanningReportService(JobService):
kind = "planning"
title = "Ticket planning"
description = (
"Builds a grouped, ordered markdown planning report of all open Gitea tickets off "
"the request path using the internal AI service (with a deterministic fallback), "
"writes the markdown artifact, and prunes it once unused for the retention window."
"Builds a complete, phased markdown implementation document of all open Gitea "
"tickets off the request path using the internal AI service (with a deterministic "
"fallback), writes the markdown artifact, and prunes it once unused for the "
"retention window. AI usage is metered into the issue tracker statistics."
)
def __init__(self) -> None:
@@ -42,8 +45,11 @@ class PlanningReportService(JobService):
raise GiteaError("Gitea integration is not configured", status=503)
issues = await self._collect_open(config)
markdown, ai_used = await generate_plan(issues, config)
usage_totals = new_usage_totals()
markdown, ai_used = await generate_plan(issues, config, usage_totals)
issue_count = len(issues)
if usage_totals["calls"]:
add_issue_usage(usage_totals)
final_name = self._final_name(job.get("preferred_name", ""), markdown)
target_dir = PLANNING_REPORTS_DIR / _directory_for(uid)
+9 -45
View File
@@ -23,7 +23,11 @@ from devplacepy.database import (
internal_gateway_key,
)
from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.openai_gateway.usage import parse_usage_headers
from devplacepy.services.openai_gateway.usage import (
accumulate_usage,
new_usage_totals,
usage_metric_cards,
)
from devplacepy.utils import generate_uid, make_combined_slug, strip_html
from devplacepy.services.audit import record as audit
@@ -209,31 +213,6 @@ def _get_ai_key() -> str:
return internal_gateway_key()
USAGE_FIELDS = (
"calls",
"prompt_tokens",
"completion_tokens",
"total_tokens",
"cost_usd",
"upstream_latency_ms",
"total_latency_ms",
)
def _new_usage_totals() -> dict:
return {field: 0 for field in USAGE_FIELDS}
def _accumulate_usage(totals: dict | None, response) -> None:
if totals is None:
return
parsed = parse_usage_headers(getattr(response, "headers", None))
if not parsed:
return
for field in USAGE_FIELDS:
totals[field] += parsed[field]
def _extract_grade(text: str) -> int | None:
match = re.search(r"\d+", text.strip())
if match:
@@ -520,7 +499,7 @@ class NewsService(BaseService):
failed_count = 0
rejected_count = 0
skipped_count = 0
usage_totals = _new_usage_totals()
usage_totals = new_usage_totals()
async with net_guard.guarded_async_client(
timeout=IMG_FETCH_TIMEOUT
@@ -623,22 +602,7 @@ class NewsService(BaseService):
)
def collect_metrics(self) -> dict:
usage = get_news_usage()
stats = [
{"label": "AI calls", "value": usage["calls"]},
{"label": "Total tokens", "value": usage["total_tokens"]},
{"label": "Prompt tokens", "value": usage["prompt_tokens"]},
{"label": "Completion tokens", "value": usage["completion_tokens"]},
{"label": "Total cost", "value": f"${usage['cost_usd']:.4f}"},
{"label": "Avg tokens/call", "value": usage["avg_tokens"]},
{"label": "Avg cost/call", "value": f"${usage['avg_cost_usd']:.6f}"},
{
"label": "Avg latency",
"value": f"{usage['avg_upstream_latency_ms']:.0f}ms",
},
{"label": "Avg tokens/sec", "value": usage["avg_tokens_per_second"]},
]
return {"stats": stats}
return {"stats": usage_metric_cards(get_news_usage())}
def _resolve_uid(self, news_table, external_id: str) -> tuple[str, bool]:
existing = news_table.find_one(external_id=external_id)
@@ -946,7 +910,7 @@ class NewsService(BaseService):
if resp.status_code != 200:
self.log(f"AI grading returned {resp.status_code}: {resp.text[:200]}")
resp.raise_for_status()
_accumulate_usage(totals, resp)
accumulate_usage(totals, resp)
result = resp.json()
text = result.get("choices", [{}])[0].get("message", {}).get("content", "")
if not text:
@@ -1000,7 +964,7 @@ class NewsService(BaseService):
f"AI formatting returned {resp.status_code}: {resp.text[:200]}"
)
resp.raise_for_status()
_accumulate_usage(totals, resp)
accumulate_usage(totals, resp)
result = resp.json()
text = result.get("choices", [{}])[0].get("message", {}).get("content", "")
text = _strip_md_fence(text or "")
@@ -284,6 +284,45 @@ def parse_usage_headers(headers) -> Optional[dict]:
}
USAGE_FIELDS = (
"calls",
"prompt_tokens",
"completion_tokens",
"total_tokens",
"cost_usd",
"upstream_latency_ms",
"total_latency_ms",
)
def new_usage_totals() -> dict:
return {field: 0 for field in USAGE_FIELDS}
def accumulate_usage(totals: Optional[dict], response) -> None:
if totals is None:
return
parsed = parse_usage_headers(getattr(response, "headers", None))
if not parsed:
return
for field in USAGE_FIELDS:
totals[field] += parsed[field]
def usage_metric_cards(usage: dict) -> list[dict]:
return [
{"label": "AI calls", "value": usage["calls"]},
{"label": "Total tokens", "value": usage["total_tokens"]},
{"label": "Prompt tokens", "value": usage["prompt_tokens"]},
{"label": "Completion tokens", "value": usage["completion_tokens"]},
{"label": "Total cost", "value": f"${usage['cost_usd']:.4f}"},
{"label": "Avg tokens/call", "value": usage["avg_tokens"]},
{"label": "Avg cost/call", "value": f"${usage['avg_cost_usd']:.6f}"},
{"label": "Avg latency", "value": f"{usage['avg_upstream_latency_ms']:.0f}ms"},
{"label": "Avg tokens/sec", "value": usage["avg_tokens_per_second"]},
]
class GatewayUsageLedger:
def record(self, raw: dict, pricing: Pricing, context_map: dict) -> Optional[dict]:
try:
@@ -50,6 +50,8 @@ export class PlanningGenerator {
poll(statusUrl, ui) {
return JobPoller.run(statusUrl, {
intervalMs: 2000,
maxAttempts: 400,
onDone: (status) => {
if (ui.status) ui.status.hidden = true;
this.renderReport(status, ui);
@@ -8,7 +8,7 @@
<div class="admin-header planning-header">
<div>
<h1>Ticket Planning</h1>
<p class="planning-subtitle">Generate a grouped, ordered planning report of every open ticket.</p>
<p class="planning-subtitle">Generate a complete, phased implementation document covering every open ticket, detailed enough to hand straight to a coding agent.</p>
</div>
{% if configured %}
<button type="button" class="btn btn-primary" data-planning-generate data-action="/issues/planning">