Initial publish v0.1.0: standalone workflow core (corpus + examples + guards)
This commit is contained in:
octopus
@@ -0,0 +1,830 @@
|
||||
import { Database } from "bun:sqlite"
|
||||
import { readdirSync, readFileSync, existsSync, statSync } from "node:fs"
|
||||
import { join, relative } from "node:path"
|
||||
import { homedir } from "node:os"
|
||||
|
||||
const DBS_DIR = join(homedir(), ".local/share/octopus")
|
||||
const LEDGER_PATH = join(DBS_DIR, "token-stage-ledger.jsonl")
|
||||
|
||||
// #2628: workspace containers (octopus-ws-*) persist their session store in
|
||||
// named volumes octopus-sessions-<id> (bind added in
|
||||
// packages/containers/src/runtime/docker.ts). Scan those alongside the host
|
||||
// dir so telemetry no longer depends on which machine/container ran a session.
|
||||
const SESSION_VOLUMES_ROOT = "/data/docker/volumes"
|
||||
|
||||
function collectSessionVolumeDbs(): string[] {
|
||||
let vols: string[] = []
|
||||
try {
|
||||
vols = readdirSync(SESSION_VOLUMES_ROOT).filter((d) => d.startsWith("octopus-sessions-"))
|
||||
} catch {
|
||||
return [] // not on the docker host (e.g. a dev workstation) — fine
|
||||
}
|
||||
const out: string[] = []
|
||||
for (const v of vols) {
|
||||
const dir = join(SESSION_VOLUMES_ROOT, v, "_data")
|
||||
try {
|
||||
for (const e of readdirSync(dir, { withFileTypes: true })) {
|
||||
if (e.isFile() && e.name.endsWith(".db") && (e.name === "octopus.db" || e.name.startsWith("octopus-"))) {
|
||||
out.push(join(dir, e.name))
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// unreadable volume — skip it
|
||||
}
|
||||
}
|
||||
return out.sort()
|
||||
}
|
||||
|
||||
function dbLabel(dbPath: string): string {
|
||||
const base = dbPath.split(/[/\\]/).pop()!.replace(".db", "")
|
||||
const vol = dbPath.match(/octopus-sessions-([a-zA-Z0-9-]+)[/\\]_data/)
|
||||
return vol ? `${base}@${vol[1]!.slice(0, 8)}` : base
|
||||
}
|
||||
|
||||
// SINCE_DAYS=<N> env var scopes queries to messages from the last N days,
|
||||
// avoiding full-table scans on multi-GB databases.
|
||||
// message.time_created is MILLISECONDS (verified: raw values ~1.78e12). Keep
|
||||
// `since` in ms — a seconds-based value is always smaller than every ms
|
||||
// timestamp, so the filter would silently match everything (#2599).
|
||||
const sinceDays = Number(process.env.SINCE_DAYS ?? "30")
|
||||
const since = sinceDays > 0 ? Date.now() - sinceDays * 86_400_000 : 0
|
||||
|
||||
// Cycle-window filters (retro #4034 quick-wins): `--since <ISO-date>` and
|
||||
// `--slug <slug>` constrain the cycle-window metrics (M2/M3/M5) to the
|
||||
// window / matching run. M1/M4 keep the SINCE_DAYS env semantics. A filter
|
||||
// that yields no data prints an explicit "no data in window" line for the
|
||||
// metric — never a silent fallback to all-time numbers.
|
||||
const argv = process.argv.slice(2)
|
||||
const arg = (name: string): string | undefined => {
|
||||
const i = argv.indexOf(`--${name}`)
|
||||
return i >= 0 ? argv[i + 1] : undefined
|
||||
}
|
||||
const sinceArg = arg("since")
|
||||
const slugArg = arg("slug")
|
||||
const windowSince = sinceArg !== undefined ? Date.parse(sinceArg) : undefined
|
||||
if (sinceArg !== undefined && Number.isNaN(windowSince)) {
|
||||
console.error(`invalid --since "${sinceArg}" — use an ISO date (e.g. 2026-09-02)`)
|
||||
process.exit(1)
|
||||
}
|
||||
const filtersActive = sinceArg !== undefined || slugArg !== undefined
|
||||
const windowOrSince = windowSince ?? since
|
||||
const m5Filter = filtersActive ? { since: windowOrSince, slug: slugArg } : undefined
|
||||
|
||||
function percentile(sorted: number[], p: number): number {
|
||||
if (sorted.length === 0) return 0
|
||||
const idx = Math.min(Math.floor((sorted.length * p) / 100), sorted.length - 1)
|
||||
return sorted[idx] ?? 0
|
||||
}
|
||||
|
||||
type AgentStats = Map<string, { msgCount: number; totalTokens: number }>
|
||||
|
||||
type SessionRow = {
|
||||
id: string
|
||||
parent_id: string | null
|
||||
slug: string | null
|
||||
title: string | null
|
||||
directory: string | null
|
||||
}
|
||||
|
||||
function loadSessionRows(db: Database): Map<string, SessionRow> | null {
|
||||
const hasTable = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='session'").get()
|
||||
if (!hasTable) return null
|
||||
const rows = db.prepare("SELECT id, parent_id, slug, title, directory FROM session").all() as SessionRow[]
|
||||
return new Map(rows.map((r) => [r.id, r]))
|
||||
}
|
||||
|
||||
// Session→run matching for --slug: a session matches when its slug, title,
|
||||
// or directory contains the run slug (workflow sessions live in worktrees
|
||||
// named after the run). A match propagates to the whole subtree (subagents),
|
||||
// so every message of the run's sessions is included.
|
||||
function slugSessionIncludeSet(sessions: Map<string, SessionRow>, slug: string): Set<string> {
|
||||
const needle = slug.toLowerCase()
|
||||
const children = new Map<string, string[]>()
|
||||
for (const r of sessions.values()) {
|
||||
if (!r.parent_id) continue
|
||||
const arr = children.get(r.parent_id) ?? []
|
||||
arr.push(r.id)
|
||||
children.set(r.parent_id, arr)
|
||||
}
|
||||
const include = new Set<string>()
|
||||
const markSubtree = (id: string) => {
|
||||
if (include.has(id)) return
|
||||
include.add(id)
|
||||
for (const c of children.get(id) ?? []) markSubtree(c)
|
||||
}
|
||||
for (const r of sessions.values()) {
|
||||
const hay = [r.slug, r.title, r.directory].filter((x): x is string => typeof x === "string")
|
||||
if (hay.some((x) => x.toLowerCase().includes(needle))) markSubtree(r.id)
|
||||
}
|
||||
return include
|
||||
}
|
||||
|
||||
function processDb(dbPath: string, since = 0, m5Filter?: { since: number; slug?: string }) {
|
||||
const dbName = dbLabel(dbPath)
|
||||
const sessions: {
|
||||
sessionId: string
|
||||
msgCount: number
|
||||
inputs: number[]
|
||||
totalInput: number
|
||||
totalCacheRead: number
|
||||
}[] = []
|
||||
const agents: AgentStats = new Map()
|
||||
|
||||
let db: Database | null = null
|
||||
try {
|
||||
db = new Database(dbPath, { readonly: true })
|
||||
db.exec("PRAGMA busy_timeout = 5000")
|
||||
} catch {
|
||||
return { dbName, sessionCount: 0, messageCount: 0, sessions, agents }
|
||||
}
|
||||
|
||||
try {
|
||||
const hasTable = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='message'").get()
|
||||
if (!hasTable) return { dbName, sessionCount: 0, messageCount: 0, sessions, agents }
|
||||
|
||||
// Single query replaces the former N+1 pattern (one query per session).
|
||||
// Grouping in JS avoids N full-table scans with json_extract.
|
||||
const rows = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT
|
||||
session_id,
|
||||
CAST(json_extract(data, '$.tokens.input') AS REAL) as inp,
|
||||
CAST(json_extract(data, '$.tokens.cache.read') AS REAL) as cr
|
||||
FROM message
|
||||
WHERE json_extract(data, '$.role') = 'assistant'
|
||||
AND json_extract(data, '$.tokens.input') IS NOT NULL
|
||||
${since > 0 ? "AND time_created >= ?" : ""}
|
||||
ORDER BY session_id, time_created
|
||||
`,
|
||||
)
|
||||
.all(...(since > 0 ? [since] : [])) as {
|
||||
session_id: string
|
||||
inp: number | null
|
||||
cr: number | null
|
||||
}[]
|
||||
|
||||
const sessionMap = new Map<string, { inputs: number[]; totalInput: number; totalCacheRead: number }>()
|
||||
|
||||
for (const r of rows) {
|
||||
const inp = Number(r.inp ?? 0)
|
||||
const cr = Number(r.cr ?? 0)
|
||||
if (inp <= 0) continue
|
||||
let s = sessionMap.get(r.session_id)
|
||||
if (!s) {
|
||||
s = { inputs: [], totalInput: 0, totalCacheRead: 0 }
|
||||
sessionMap.set(r.session_id, s)
|
||||
}
|
||||
s.inputs.push(inp)
|
||||
s.totalInput += inp
|
||||
s.totalCacheRead += cr
|
||||
}
|
||||
|
||||
for (const [sessionId, s] of sessionMap) {
|
||||
if (s.inputs.length === 0) continue
|
||||
sessions.push({
|
||||
sessionId,
|
||||
msgCount: s.inputs.length,
|
||||
inputs: s.inputs,
|
||||
totalInput: s.totalInput,
|
||||
totalCacheRead: s.totalCacheRead,
|
||||
})
|
||||
}
|
||||
|
||||
if (m5Filter) {
|
||||
let include: Set<string> | undefined
|
||||
if (m5Filter.slug !== undefined) {
|
||||
const sessions = loadSessionRows(db)
|
||||
include = sessions ? slugSessionIncludeSet(sessions, m5Filter.slug) : new Set<string>()
|
||||
}
|
||||
// include.size === 0 (or a slug-less filter) → contribute nothing;
|
||||
// the merged M5 output prints the explicit "no data in window" line.
|
||||
if (include === undefined || include.size > 0) {
|
||||
const rows = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT
|
||||
session_id,
|
||||
COALESCE(json_extract(data, '$.agent'), 'unknown') as agent,
|
||||
CAST(json_extract(data, '$.tokens.input') AS REAL) as inp,
|
||||
CAST(json_extract(data, '$.tokens.output') AS REAL) as outp,
|
||||
CAST(json_extract(data, '$.tokens.reasoning') AS REAL) as rea
|
||||
FROM message
|
||||
WHERE json_extract(data, '$.role') = 'assistant'
|
||||
AND json_extract(data, '$.tokens.input') IS NOT NULL
|
||||
${m5Filter.since > 0 ? "AND time_created >= ?" : ""}
|
||||
`,
|
||||
)
|
||||
.all(...(m5Filter.since > 0 ? [m5Filter.since] : [])) as {
|
||||
session_id: string
|
||||
agent: string
|
||||
inp: number | null
|
||||
outp: number | null
|
||||
rea: number | null
|
||||
}[]
|
||||
for (const r of rows) {
|
||||
if (include !== undefined && !include.has(r.session_id)) continue
|
||||
const cur = agents.get(r.agent) ?? { msgCount: 0, totalTokens: 0 }
|
||||
cur.msgCount++
|
||||
cur.totalTokens += Number(r.inp ?? 0) + Number(r.outp ?? 0) + Number(r.rea ?? 0)
|
||||
agents.set(r.agent, cur)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const agentRows = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT
|
||||
COALESCE(json_extract(data, '$.agent'), 'unknown') as agent,
|
||||
COUNT(*) as msg_count,
|
||||
CAST(TOTAL(json_extract(data, '$.tokens.input')) AS REAL) as inp,
|
||||
CAST(TOTAL(json_extract(data, '$.tokens.output')) AS REAL) as outp,
|
||||
CAST(TOTAL(json_extract(data, '$.tokens.reasoning')) AS REAL) as rea
|
||||
FROM message
|
||||
WHERE json_extract(data, '$.role') = 'assistant'
|
||||
AND json_extract(data, '$.tokens.input') IS NOT NULL
|
||||
${since > 0 ? "AND time_created >= ?" : ""}
|
||||
GROUP BY agent
|
||||
`,
|
||||
)
|
||||
.all(...(since > 0 ? [since] : [])) as {
|
||||
agent: string
|
||||
msg_count: number
|
||||
inp: number
|
||||
outp: number
|
||||
rea: number
|
||||
}[]
|
||||
|
||||
for (const r of agentRows) {
|
||||
agents.set(r.agent, {
|
||||
msgCount: Number(r.msg_count ?? 0),
|
||||
totalTokens: Number(r.inp ?? 0) + Number(r.outp ?? 0) + Number(r.rea ?? 0),
|
||||
})
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
|
||||
const messageCount = sessions.reduce((s, se) => s + se.msgCount, 0)
|
||||
return { dbName, sessionCount: sessions.length, messageCount, sessions, agents }
|
||||
}
|
||||
|
||||
function dbSummary(db: ReturnType<typeof processDb>) {
|
||||
if (db.sessionCount === 0) return null
|
||||
return ` ${db.dbName}: ${db.sessionCount} sessions, ${db.messageCount.toLocaleString()} messages`
|
||||
}
|
||||
|
||||
function computeSessionMedians(dbs: ReturnType<typeof processDb>[]) {
|
||||
const medians: number[] = []
|
||||
for (const db of dbs) {
|
||||
for (const s of db.sessions) {
|
||||
const sorted = [...s.inputs].sort((a, b) => a - b)
|
||||
medians.push(percentile(sorted, 50))
|
||||
}
|
||||
}
|
||||
return medians
|
||||
}
|
||||
|
||||
function computeSessionMaxes(dbs: ReturnType<typeof processDb>[]) {
|
||||
const maxes: number[] = []
|
||||
for (const db of dbs) {
|
||||
for (const s of db.sessions) {
|
||||
maxes.push(Math.max(...s.inputs))
|
||||
}
|
||||
}
|
||||
return maxes
|
||||
}
|
||||
|
||||
function computeSessionP90s(dbs: ReturnType<typeof processDb>[]) {
|
||||
const p90s: number[] = []
|
||||
for (const db of dbs) {
|
||||
for (const s of db.sessions) {
|
||||
const sorted = [...s.inputs].sort((a, b) => a - b)
|
||||
p90s.push(percentile(sorted, 90))
|
||||
}
|
||||
}
|
||||
return p90s
|
||||
}
|
||||
|
||||
function computeOverallCacheRatio(dbs: ReturnType<typeof processDb>[]) {
|
||||
let totalInput = 0
|
||||
let totalCacheRead = 0
|
||||
for (const db of dbs) {
|
||||
for (const s of db.sessions) {
|
||||
totalInput += s.totalInput
|
||||
totalCacheRead += s.totalCacheRead
|
||||
}
|
||||
}
|
||||
return totalInput > 0 ? totalCacheRead / totalInput : 0
|
||||
}
|
||||
|
||||
function trafficLight(ratio: number): string {
|
||||
if (ratio > 10) return "🟢"
|
||||
if (ratio >= 3) return "🟡"
|
||||
return "🔴"
|
||||
}
|
||||
|
||||
const dbFiles = [
|
||||
...readdirSync(DBS_DIR, { withFileTypes: true })
|
||||
.filter((e) => e.isFile() && e.name.endsWith(".db") && (e.name === "octopus.db" || e.name.startsWith("octopus-")))
|
||||
.map((e) => join(DBS_DIR, e.name)),
|
||||
...collectSessionVolumeDbs(), // #2628
|
||||
].sort()
|
||||
|
||||
for (const f of dbFiles) {
|
||||
const size = statSync(f).size
|
||||
if (size > 1_000_000_000 && since === 0) {
|
||||
console.error(
|
||||
`[WARN] ${f.split("/").pop()} is ${(size / 1e9).toFixed(1)} GB — query may be slow. Set SINCE_DAYS=<N> to scope to recent sessions.`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
const allDbs = dbFiles.map((p) => processDb(p, since, m5Filter))
|
||||
|
||||
console.log("# Token Telemetry: Context Hygiene (M4)")
|
||||
console.log()
|
||||
if (filtersActive) {
|
||||
const parts = [sinceArg ? `--since ${sinceArg}` : "", slugArg ? `--slug ${slugArg}` : ""].filter(Boolean)
|
||||
console.log(`Cycle-window filters active (${parts.join(" ")}): M2/M3/M5 constrained to the window; M1/M4 unchanged`)
|
||||
console.log()
|
||||
}
|
||||
|
||||
console.log("## Per-Database Summaries")
|
||||
console.log()
|
||||
for (const db of allDbs) {
|
||||
const s = dbSummary(db)
|
||||
if (s) console.log(s)
|
||||
}
|
||||
|
||||
const totalSessions = allDbs.reduce((s, d) => s + d.sessionCount, 0)
|
||||
const totalMessages = allDbs.reduce((s, d) => s + d.messageCount, 0)
|
||||
console.log()
|
||||
console.log(
|
||||
`Total across ${allDbs.filter((d) => d.sessionCount > 0).length} databases: ${totalSessions} sessions, ${totalMessages.toLocaleString()} messages`,
|
||||
)
|
||||
|
||||
const sessionMedians = computeSessionMedians(allDbs)
|
||||
const sessionP90s = computeSessionP90s(allDbs)
|
||||
const sessionMaxes = computeSessionMaxes(allDbs)
|
||||
const cacheRatio = computeOverallCacheRatio(allDbs)
|
||||
|
||||
const sortedMedians = [...sessionMedians].sort((a, b) => a - b)
|
||||
const sortedP90s = [...sessionP90s].sort((a, b) => a - b)
|
||||
const sortedMaxes = [...sessionMaxes].sort((a, b) => a - b)
|
||||
|
||||
console.log()
|
||||
console.log("## Aggregate Input Token Stats (per-session metrics)")
|
||||
console.log()
|
||||
console.log("| Metric | p50 | p90 | max |")
|
||||
console.log("| ------ | --- | --- | --- |")
|
||||
console.log(
|
||||
`| Per-session median input | ${percentile(sortedMedians, 50).toLocaleString()} | ${percentile(sortedMedians, 90).toLocaleString()} | ${percentile(sortedMedians, 100).toLocaleString()} |`,
|
||||
)
|
||||
console.log(
|
||||
`| Per-session p90 input | ${percentile(sortedP90s, 50).toLocaleString()} | ${percentile(sortedP90s, 90).toLocaleString()} | ${percentile(sortedP90s, 100).toLocaleString()} |`,
|
||||
)
|
||||
console.log(
|
||||
`| Per-session max input | ${percentile(sortedMaxes, 50).toLocaleString()} | ${percentile(sortedMaxes, 90).toLocaleString()} | ${percentile(sortedMaxes, 100).toLocaleString()} |`,
|
||||
)
|
||||
|
||||
console.log()
|
||||
const ratioLabel = cacheRatio >= 1 ? `${cacheRatio.toFixed(1)}:1` : `1:${(1 / cacheRatio).toFixed(1)}`
|
||||
const light = trafficLight(cacheRatio)
|
||||
console.log(`## Cache Read / Input Ratio: ${ratioLabel} ${light}`)
|
||||
console.log()
|
||||
|
||||
const desc =
|
||||
cacheRatio > 10
|
||||
? "Excellent — context reuse is very high, indicating effective caching"
|
||||
: cacheRatio >= 3
|
||||
? "Moderate — reasonable cache hits, room for improvement"
|
||||
: "Low — consider strategies to increase context cache reuse"
|
||||
console.log(` ${desc}`)
|
||||
|
||||
const mergedAgents: Map<string, { msgCount: number; totalTokens: number }> = new Map()
|
||||
for (const db of allDbs) {
|
||||
for (const [agent, stats] of db.agents) {
|
||||
const existing = mergedAgents.get(agent)
|
||||
if (existing) {
|
||||
existing.msgCount += stats.msgCount
|
||||
existing.totalTokens += stats.totalTokens
|
||||
} else {
|
||||
mergedAgents.set(agent, { ...stats })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let exploreTokens = 0
|
||||
let workerTokens = 0
|
||||
for (const [agent, stats] of mergedAgents) {
|
||||
const lower = agent.toLowerCase()
|
||||
if (lower.includes("explorer")) exploreTokens += stats.totalTokens
|
||||
else if (lower.includes("worker")) workerTokens += stats.totalTokens
|
||||
}
|
||||
|
||||
const m5Ratio = workerTokens > 0 ? exploreTokens / workerTokens : 0
|
||||
|
||||
function m5TrafficLight(ratio: number): string {
|
||||
if (ratio > 2) return "🟢"
|
||||
if (ratio >= 1) return "🟡"
|
||||
return "🔴"
|
||||
}
|
||||
|
||||
console.log()
|
||||
console.log("## Explore / Execute Ratio (M5)")
|
||||
console.log()
|
||||
console.log("| Agent | Messages | Total tokens |")
|
||||
console.log("| --------- | -------- | ------------ |")
|
||||
|
||||
const sortedAgents = [...mergedAgents.entries()].sort((a, b) => b[1].totalTokens - a[1].totalTokens)
|
||||
for (const [agent, stats] of sortedAgents) {
|
||||
console.log(
|
||||
`| ${agent.padEnd(9)} | ${stats.msgCount.toLocaleString().padStart(7)} | ${stats.totalTokens.toLocaleString().padStart(12)} |`,
|
||||
)
|
||||
}
|
||||
|
||||
console.log()
|
||||
if (filtersActive && mergedAgents.size === 0) {
|
||||
console.log("[NOTE: no data in window for M5 — no messages match the requested window/slug]")
|
||||
} else {
|
||||
const m5Light = m5TrafficLight(m5Ratio)
|
||||
console.log(`Explore/Execute: ${m5Ratio.toFixed(2)}:1 ${m5Light}`)
|
||||
console.log()
|
||||
|
||||
const m5Desc =
|
||||
m5Ratio > 2
|
||||
? "Explorer-heavy — exploration dominates execution, good for discovery but may need more synthesis"
|
||||
: m5Ratio >= 1
|
||||
? "Balanced — reasonable split between exploration and execution"
|
||||
: "Execution-heavy — workers are spending tokens on discovery work that explorers should handle"
|
||||
console.log(` ${m5Desc}`)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Compactor activity — compact-frequency proxy (#2601 pilot data).
|
||||
// The compactor agent runs once per agent-initiated compaction, so its
|
||||
// message count in the window approximates how often compaction fired.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
console.log()
|
||||
console.log("## Compactor Activity (#2601 compact-frequency proxy)")
|
||||
console.log()
|
||||
const compactor = mergedAgents.get("compactor")
|
||||
if (compactor) {
|
||||
console.log(
|
||||
`compactor: ${compactor.msgCount.toLocaleString()} messages, ${compactor.totalTokens.toLocaleString()} tokens in window`,
|
||||
)
|
||||
console.log(" (per-run distribution = the #2601 pilot metric; rising zero-compact")
|
||||
console.log(" share for short runs (bugfix / DAG task) with no late-stage degradation retires the pilot gate)")
|
||||
} else {
|
||||
console.log("[NOTE: no compactor messages in window — zero agent-initiated compactions recorded]")
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// M2 — Stage token distribution (ledger-gated).
|
||||
// Reads token-stage-ledger.jsonl (written by .octopus/plugin/token-stage-ledger.ts),
|
||||
// reconstructs a per-root-session stage timeline, and attributes every
|
||||
// assistant message's tokens to the stage that was active when the message
|
||||
// was created. Skips cleanly when no ledger exists.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type LedgerEntry = { sessionID: string; stage: string; t: number }
|
||||
|
||||
function loadLedger(): Map<string, { stage: string; t: number }[]> | null {
|
||||
if (!existsSync(LEDGER_PATH)) return null
|
||||
const bySession = new Map<string, { stage: string; t: number }[]>()
|
||||
let any = false
|
||||
for (const line of readFileSync(LEDGER_PATH, "utf8").split("\n")) {
|
||||
const trimmed = line.trim()
|
||||
if (!trimmed) continue
|
||||
try {
|
||||
const e = JSON.parse(trimmed) as LedgerEntry
|
||||
const arr = bySession.get(e.sessionID) ?? []
|
||||
arr.push({ stage: e.stage, t: e.t })
|
||||
bySession.set(e.sessionID, arr)
|
||||
any = true
|
||||
} catch {
|
||||
// skip malformed lines
|
||||
}
|
||||
}
|
||||
if (!any) return null
|
||||
for (const arr of bySession.values()) arr.sort((a, b) => a.t - b.t)
|
||||
return bySession
|
||||
}
|
||||
|
||||
function stageTokensForDb(
|
||||
dbPath: string,
|
||||
ledger: Map<string, { stage: string; t: number }[]>,
|
||||
since = 0,
|
||||
slug?: string,
|
||||
): Map<string, number> | null {
|
||||
let db: Database | null = null
|
||||
try {
|
||||
db = new Database(dbPath, { readonly: true })
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
try {
|
||||
const sessions = loadSessionRows(db)
|
||||
if (!sessions) return null
|
||||
|
||||
// timelines keyed by root sessions present in this db
|
||||
const timelines = new Map<string, { stage: string; t: number }[]>()
|
||||
for (const [sid, entries] of ledger) {
|
||||
if (sessions.has(sid)) timelines.set(sid, entries)
|
||||
}
|
||||
if (timelines.size === 0) return null
|
||||
|
||||
let include: Set<string> | undefined
|
||||
if (slug !== undefined) {
|
||||
include = slugSessionIncludeSet(sessions, slug)
|
||||
if (include.size === 0) return null // no session matches the slug
|
||||
}
|
||||
|
||||
const rootOf = (id: string): string => {
|
||||
let cur = id
|
||||
let guard = 0
|
||||
while (guard++ < 100) {
|
||||
const parent = sessions.get(cur)?.parent_id
|
||||
if (!parent) break
|
||||
cur = parent
|
||||
}
|
||||
return cur
|
||||
}
|
||||
const stageAt = (rootId: string, time: number): string | null => {
|
||||
const tl = timelines.get(rootId)
|
||||
if (!tl) return null
|
||||
let stage: string | null = null
|
||||
for (const e of tl) {
|
||||
if (e.t <= time) stage = e.stage
|
||||
else break
|
||||
}
|
||||
return stage
|
||||
}
|
||||
|
||||
const byStage = new Map<string, number>()
|
||||
const rows = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT session_id, time_created,
|
||||
CAST(json_extract(data, '$.tokens.input') AS REAL) as inp,
|
||||
CAST(json_extract(data, '$.tokens.output') AS REAL) as outp,
|
||||
CAST(json_extract(data, '$.tokens.reasoning') AS REAL) as rea
|
||||
FROM message
|
||||
WHERE json_extract(data, '$.role') = 'assistant'
|
||||
AND json_extract(data, '$.tokens.input') IS NOT NULL
|
||||
${since > 0 ? "AND time_created >= ?" : ""}
|
||||
`,
|
||||
)
|
||||
.all(...(since > 0 ? [since] : [])) as {
|
||||
session_id: string
|
||||
time_created: number
|
||||
inp: number | null
|
||||
outp: number | null
|
||||
rea: number | null
|
||||
}[]
|
||||
|
||||
for (const r of rows) {
|
||||
if (include !== undefined && !include.has(r.session_id)) continue
|
||||
const stage = stageAt(rootOf(r.session_id), r.time_created)
|
||||
if (!stage) continue
|
||||
const tokens = Number(r.inp ?? 0) + Number(r.outp ?? 0) + Number(r.rea ?? 0)
|
||||
byStage.set(stage, (byStage.get(stage) ?? 0) + tokens)
|
||||
}
|
||||
return byStage
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// M3 — Review rework fraction (#2591).
|
||||
// Two sources, merged with dedup by review identity `{slug}/reviews/{stage}`
|
||||
// (the ACTIVE status.json wins when both exist — it is canonical):
|
||||
// 1. ACTIVE runs — status.json under the Tier 1 location
|
||||
// .octopus/runs/{slug}/reviews/{stage}/status.json (reads history[],
|
||||
// legacy alias rounds[], current_round) and the legacy
|
||||
// .artifacts/**/reviews/*/status.json tree. In-flight runs only: the
|
||||
// active workspace is deleted at archive-at-close, so this source alone
|
||||
// structurally empties as runs close.
|
||||
// 2. ARCHIVED runs — the committed archive bundle
|
||||
// .octopus/runs/archive/{slug}.json. Bundles store digests, not
|
||||
// status.json content, so the per-review round count is reconstructed
|
||||
// from the documented Tier 1 layout `reviews/{stage}/round{N}/…`
|
||||
// (templates/runs-layout.md) by counting distinct roundN path segments
|
||||
// per stage across index.artifacts[].path. This archive source is what
|
||||
// makes M3 durable.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type ReviewRounds = Map<string, number> // `${slug}/reviews/${stage}` -> rounds
|
||||
|
||||
function collectActiveReviewRounds(roots: string[]): { rounds: ReviewRounds; startedMs: Map<string, number> } {
|
||||
const out: ReviewRounds = new Map()
|
||||
const startedMs = new Map<string, number>()
|
||||
const walk = (dir: string, top: string) => {
|
||||
let entries: ReturnType<typeof readdirSync>
|
||||
try {
|
||||
entries = readdirSync(dir, { withFileTypes: true })
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
for (const e of entries) {
|
||||
if (e.name === "_archive" || e.name === "archive") continue
|
||||
const full = join(dir, e.name)
|
||||
if (e.isDirectory()) walk(full, top)
|
||||
else if (e.name === "status.json" && dir.includes("/reviews/")) {
|
||||
try {
|
||||
const data = JSON.parse(readFileSync(full, "utf8")) as {
|
||||
rounds?: unknown[]
|
||||
history?: unknown[]
|
||||
current_round?: number
|
||||
started_at?: string
|
||||
}
|
||||
// Canonical field is `history[]` (per review-status.schema.json);
|
||||
// `rounds[]` is a legacy alias that maps to it. Prefer the array
|
||||
// forms; fall back to current_round.
|
||||
let rounds = 0
|
||||
if (Array.isArray(data.rounds) && data.rounds.length > 0) rounds = data.rounds.length
|
||||
else if (Array.isArray(data.history) && data.history.length > 0) rounds = data.history.length
|
||||
else if (typeof data.current_round === "number" && data.current_round > 0) rounds = data.current_round
|
||||
if (rounds > 0) {
|
||||
const key = relative(top, dir).split("\\").join("/")
|
||||
out.set(key, rounds)
|
||||
if (typeof data.started_at === "string") {
|
||||
const t = Date.parse(data.started_at)
|
||||
if (!Number.isNaN(t)) startedMs.set(key, t)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// skip unreadable / malformed status files
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const root of roots) walk(root, root)
|
||||
return { rounds: out, startedMs }
|
||||
}
|
||||
|
||||
function collectBundleReviewRounds(archiveDir: string): { rounds: ReviewRounds; closedMs: Map<string, number> } {
|
||||
let files: string[] = []
|
||||
try {
|
||||
files = readdirSync(archiveDir).filter((f) => f.endsWith(".json"))
|
||||
} catch {
|
||||
return { rounds: new Map(), closedMs: new Map() } // no archive dir (e.g. a fresh checkout) — fine
|
||||
}
|
||||
const out: ReviewRounds = new Map()
|
||||
const closedMs = new Map<string, number>()
|
||||
for (const f of files) {
|
||||
try {
|
||||
const bundle = JSON.parse(readFileSync(join(archiveDir, f), "utf8")) as {
|
||||
index?: { artifacts?: { path?: unknown }[] }
|
||||
meta?: { closed_at?: string; updated_at?: string; created_at?: string }
|
||||
}
|
||||
// Identity = the bundle filename stem (= the archived run's workspace
|
||||
// dir name). meta.slug is NOT unique — epic task-node bundles carry the
|
||||
// parent epic slug while filenames stay per-node.
|
||||
const slug = f.replace(/\.json$/, "")
|
||||
const closedRaw = bundle.meta?.closed_at ?? bundle.meta?.updated_at ?? bundle.meta?.created_at
|
||||
// Distinct roundN segments per review stage across artifact paths
|
||||
// (paths may or may not carry the slug prefix — match the segment).
|
||||
const byStage = new Map<string, Set<string>>()
|
||||
for (const a of bundle.index?.artifacts ?? []) {
|
||||
if (typeof a?.path !== "string") continue
|
||||
const hit = a.path.match(/reviews\/([^/]+)\/(round\d+)\//)
|
||||
if (!hit?.[1] || !hit[2]) continue
|
||||
const set = byStage.get(hit[1]) ?? new Set<string>()
|
||||
set.add(hit[2])
|
||||
byStage.set(hit[1], set)
|
||||
}
|
||||
for (const [stage, rounds] of byStage) {
|
||||
if (rounds.size > 0) out.set(`${slug}/reviews/${stage}`, rounds.size)
|
||||
if (closedRaw !== undefined) {
|
||||
const t = Date.parse(closedRaw)
|
||||
if (!Number.isNaN(t)) closedMs.set(`${slug}/reviews/${stage}`, t)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// skip unreadable / malformed bundles
|
||||
}
|
||||
}
|
||||
return { rounds: out, closedMs }
|
||||
}
|
||||
|
||||
// --- M2 output ---
|
||||
const ledger = loadLedger()
|
||||
console.log()
|
||||
console.log("## Stage Token Distribution (M2)")
|
||||
console.log()
|
||||
if (!ledger) {
|
||||
console.log("[NOTE: token-stage-ledger.jsonl absent — M2 skipped]")
|
||||
console.log(" (enable the .octopus/plugin/token-stage-ledger plugin to populate)")
|
||||
} else {
|
||||
const mergedStages = new Map<string, number>()
|
||||
for (const dbPath of dbFiles) {
|
||||
const byStage = stageTokensForDb(dbPath, ledger, windowOrSince, slugArg)
|
||||
if (!byStage) continue
|
||||
for (const [stage, tokens] of byStage) mergedStages.set(stage, (mergedStages.get(stage) ?? 0) + tokens)
|
||||
}
|
||||
const grandTotal = [...mergedStages.values()].reduce((a, b) => a + b, 0)
|
||||
if (grandTotal === 0) {
|
||||
if (filtersActive) {
|
||||
console.log("[NOTE: no data in window for M2 — no attributed tokens match the requested window/slug]")
|
||||
} else {
|
||||
console.log("[NOTE: ledger present but no sessions matched — M2 has no attributed data yet]")
|
||||
}
|
||||
} else {
|
||||
const sortedStages = [...mergedStages.entries()].sort((a, b) => b[1] - a[1])
|
||||
console.log("| Stage | Tokens | Share |")
|
||||
console.log("| ------------- | ------ | ----- |")
|
||||
for (const [stage, tokens] of sortedStages) {
|
||||
const pct = ((tokens / grandTotal) * 100).toFixed(1)
|
||||
console.log(`| ${stage.padEnd(13)} | ${tokens.toLocaleString().padStart(13)} | ${pct.padStart(5)}% |`)
|
||||
}
|
||||
const reviewTokens = mergedStages.get("review") ?? 0
|
||||
const reviewShare = (reviewTokens / grandTotal) * 100
|
||||
const m2Light = reviewShare > 60 ? "🔴" : reviewShare >= 35 ? "🟡" : "🟢"
|
||||
console.log()
|
||||
console.log(`Review-stage share: ${reviewShare.toFixed(1)}% ${m2Light}`)
|
||||
console.log(
|
||||
` ${
|
||||
reviewShare > 60
|
||||
? "Review dominates token spend — possible over-reviewing"
|
||||
: reviewShare >= 35
|
||||
? "Moderate review spend"
|
||||
: "Review spend is proportionate"
|
||||
}`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// --- M3 output ---
|
||||
const runsDir = join(process.cwd(), ".octopus", "runs")
|
||||
const artifactsDir = join(process.cwd(), ".artifacts")
|
||||
const archiveDir = join(runsDir, "archive")
|
||||
const activeRoots = [runsDir, artifactsDir].filter((d) => existsSync(d))
|
||||
console.log()
|
||||
console.log("## Review Rework (M3)")
|
||||
console.log()
|
||||
const active = collectActiveReviewRounds(activeRoots)
|
||||
const archived = collectBundleReviewRounds(archiveDir)
|
||||
const activeRounds = active.rounds
|
||||
const archivedRounds = archived.rounds
|
||||
const mergedRounds: ReviewRounds = new Map(activeRounds)
|
||||
let archivedOnly = 0
|
||||
for (const [key, rounds] of archivedRounds) {
|
||||
if (mergedRounds.has(key)) continue // active status.json is canonical
|
||||
mergedRounds.set(key, rounds)
|
||||
archivedOnly++
|
||||
}
|
||||
// Cycle-window filters: constrain to reviews whose run identity contains the
|
||||
// slug and whose start (active) / close (archived) time falls in the window.
|
||||
// Reviews without a parseable timestamp are excluded when a window is set —
|
||||
// strict, so filtered numbers never silently fall back to all-time totals.
|
||||
const windowRounds: ReviewRounds = new Map()
|
||||
for (const [key, rounds] of mergedRounds) {
|
||||
const identity = key.split("/reviews/")[0] ?? key
|
||||
if (slugArg !== undefined && !identity.includes(slugArg)) continue
|
||||
if (windowSince !== undefined) {
|
||||
const t = active.startedMs.get(key) ?? archived.closedMs.get(key)
|
||||
if (t === undefined || t < windowSince) continue
|
||||
}
|
||||
windowRounds.set(key, rounds)
|
||||
}
|
||||
if (mergedRounds.size === 0) {
|
||||
console.log(
|
||||
activeRoots.length === 0 && archivedRounds.size === 0
|
||||
? "[NOTE: no .octopus/runs or .artifacts directory in cwd — M3 skipped]"
|
||||
: "[NOTE: no review rounds found (active status.json or archive bundles) — M3 skipped]",
|
||||
)
|
||||
} else if (filtersActive && windowRounds.size === 0) {
|
||||
console.log("[NOTE: no data in window for M3 — no reviews match the requested window/slug]")
|
||||
} else {
|
||||
const roundsMap = filtersActive ? windowRounds : mergedRounds
|
||||
const reviews = roundsMap.size
|
||||
const totalRounds = [...roundsMap.values()].reduce((a, b) => a + b, 0)
|
||||
const reworkRounds = [...roundsMap.values()].reduce((a, b) => a + (b - 1), 0)
|
||||
const nonFirstPass = [...roundsMap.values()].filter((r) => r > 1).length
|
||||
const fraction = totalRounds > 0 ? reworkRounds / totalRounds : 0
|
||||
const nonFirstPct = (nonFirstPass / reviews) * 100
|
||||
const m3Light = fraction > 0.3 ? "🔴" : fraction >= 0.15 ? "🟡" : "🟢"
|
||||
if (filtersActive)
|
||||
console.log(`Window filter: ${windowRounds.size}/${mergedRounds.size} reviews match (--since/--slug)`)
|
||||
console.log(`Reviews: ${reviews} | total rounds: ${totalRounds} | rework rounds: ${reworkRounds}`)
|
||||
console.log(
|
||||
`Sources: ${activeRounds.size} active status.json + ${archivedOnly} archive bundles (dedup by slug+stage)`,
|
||||
)
|
||||
console.log(`Non-first-pass reviews: ${nonFirstPass}/${reviews} (${nonFirstPct.toFixed(0)}%)`)
|
||||
console.log()
|
||||
console.log(`Rework fraction: ${(fraction * 100).toFixed(1)}% ${m3Light}`)
|
||||
console.log(
|
||||
` ${
|
||||
fraction > 0.3
|
||||
? "High rework — review findings not actionable or design unclear"
|
||||
: fraction >= 0.15
|
||||
? "Moderate rework — some review churn"
|
||||
: "Low rework — reviews converge efficiently"
|
||||
}`,
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user