diff --git a/.github/workflows/performance-report.yml b/.github/workflows/performance-report.yml new file mode 100644 index 000000000..e867e6663 --- /dev/null +++ b/.github/workflows/performance-report.yml @@ -0,0 +1,158 @@ +name: Update PR performance comment +on: + workflow_call: + inputs: + section: + required: true + type: string + artifact-name: + required: true + type: string + measurement-result: + required: true + type: string +permissions: + contents: read + pull-requests: write +jobs: + publish: + name: Update existing performance comment + if: >- + github.event_name == 'pull_request' && + github.event.pull_request.head.repo.full_name == github.repository + runs-on: ubuntu-24.04 + timeout-minutes: 5 + # Both performance workflows read and update the same comment. Serialize the + # entire job so each merge sees the other publisher's completed update. + concurrency: + group: pr-performance-comment-${{ github.repository }}-${{ github.event.pull_request.number }} + cancel-in-progress: false + steps: + - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + id: download + continue-on-error: true + with: + name: ${{ inputs.artifact-name }} + path: report + # This write-enabled job never checks out the PR or executes artifact code. + - uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 + env: + REPORT_SECTION: ${{ inputs.section }} + MEASUREMENT_RESULT: ${{ inputs.measurement-result }} + DOWNLOAD_OUTCOME: ${{ steps.download.outcome }} + with: + script: | + const fs = require('fs'); + const marker = ''; + const legacyToken = ''; + const sections = ['token-usage', 'reader-api']; + const titles = {'token-usage': 'Token usage benchmark', 'reader-api': 'Reader API performance'}; + const section = process.env.REPORT_SECTION; + if (!sections.includes(section)) throw new Error('Unknown performance report section'); + const pr = context.payload.pull_request; + const sha = pr.head.sha; + const run = String(context.runId); + const attempt = process.env.GITHUB_RUN_ATTEMPT; + if (!/^[a-f0-9]{40}$/.test(sha) || !/^[1-9]\d*$/.test(run) || !/^[1-9]\d*$/.test(attempt)) { + throw new Error('Invalid performance report provenance'); + } + const current = await github.rest.pulls.get({...context.repo, pull_number: pr.number}); + if (current.data.head.sha !== sha) { + core.info('Skipping results for a superseded PR head.'); + return; + } + const url = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${run}`; + let summary; + let missing = false; + try { + if (process.env.DOWNLOAD_OUTCOME !== 'success') throw new Error('Artifact unavailable'); + const stat = fs.lstatSync('report/summary.md'); + if (!stat.isFile() || stat.size > 25000) throw new Error('Invalid summary file'); + summary = fs.readFileSync('report/summary.md', 'utf8'); + const valid = section === 'token-usage' + ? /^\*\*(PASS|FAIL|INCOMPLETE) — Token usage benchmark:\*\*/.test(summary) + : /^(# PR reader API performance|\*\*ERROR\*\*)/.test(summary); + if (!valid) throw new Error('Invalid summary verdict'); + } catch { + missing = true; + summary = `**INCOMPLETE — ${titles[section]}:** no readable results artifact was produced. Inspect the job logs; no performance pass can be claimed.`; + } + const outcome = process.env.MEASUREMENT_RESULT; + if (outcome !== 'success') { + const verdict = outcome === 'failure' ? 'FAIL' : 'INCOMPLETE'; + if (section === 'token-usage') { + summary = summary.replace(/^\*\*(PASS|FAIL|INCOMPLETE)/, `**${verdict}`); + } else { + summary = `**${verdict} — Reader API performance:** the comparison job did not complete successfully.\n\n${summary}`; + } + summary += `\n\nMeasurement job outcome: **${['failure', 'cancelled', 'skipped'].includes(outcome) ? outcome : 'unknown'}**. The run did not complete successfully.`; + } + // Artifact text is inert Markdown. Neutralize HTML comment delimiters + // so it cannot forge the publisher's section/provenance boundaries. + summary = summary.replace(/\n([\s\S]*?)\n$/gm; + const matches = [...comment.body.matchAll(expression)]; + if (matches.length) { + for (const match of matches) { + preserve(match[1], {head: match[2], run: match[3], attempt: match[4], text: match[5]}); + } + continue; + } + // Adopt both older standalone performance comment formats once, + // keeping their original content and provenance during migration. + const key = comment.body.startsWith(legacyToken) ? 'token-usage' : 'reader-api'; + const oldHead = comment.body.match(/(?:Compared head:|→ PR)\s*`?([a-f0-9]{40})`?/i)?.[1] ?? 'unknown'; + const tokenRun = comment.body.match(//); + const oldRun = tokenRun?.[1] ?? [...comment.body.matchAll(/\/actions\/runs\/([1-9]\d*)/g)].at(-1)?.[1] ?? '0'; + const oldAttempt = tokenRun?.[2] ?? [...comment.body.matchAll(/\battempt\s+([1-9]\d*)\b/gi)].at(-1)?.[1] ?? '1'; + const text = comment.body.replace(/^\n/gm, '').trim().replace(/\n${report.text}\n`; + }); + const body = `${marker}\n# PR performance results\n\nCurrent PR head: \`${sha}\`\n\n${rendered.join('\n\n')}`; + if (Buffer.byteLength(body, 'utf8') > 60000) { + core.setFailed('Combined performance report exceeds the comment limit; existing results were preserved.'); + return; + } + const latest = await github.rest.pulls.get({...context.repo, pull_number: pr.number}); + if (latest.data.head.sha !== sha) return; + if (previous.length) { + await github.rest.issues.updateComment({...context.repo, comment_id: previous[0].id, body}); + // Remove only duplicate, explicitly marked Actions performance + // comments after their sections have been saved in the oldest one. + for (const duplicate of previous.slice(1)) { + await github.rest.issues.deleteComment({...context.repo, comment_id: duplicate.id}); + } + } else { + await github.rest.issues.createComment({...context.repo, issue_number: pr.number, body}); + } + if (missing) core.setFailed('Published incomplete results: the summary artifact was unavailable or invalid.'); diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml index f30ceea08..d01a3bbe4 100644 --- a/.github/workflows/performance.yml +++ b/.github/workflows/performance.yml @@ -153,30 +153,11 @@ jobs: always() && !cancelled() && github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository && github.actor != 'renovate[bot]' - runs-on: ubuntu-24.04 + uses: ./.github/workflows/performance-report.yml permissions: contents: read pull-requests: write - steps: - - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 - with: - name: pr-performance-${{ github.run_attempt }} - path: report - - uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 - with: - script: | - const fs = require('fs'); - const marker = ''; - const pr = context.payload.pull_request; - const current = await github.rest.pulls.get({...context.repo, pull_number: pr.number}); - if (current.data.head.sha !== pr.head.sha) return; - const url = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`; - const summary = fs.readFileSync('report/summary.md', 'utf8').slice(0, 50000); - const body = `${marker}\n${summary}\n[Full HTML/CSV reports and logs](${url}) · attempt ${process.env.GITHUB_RUN_ATTEMPT}`; - const comments = await github.paginate(github.rest.issues.listComments, {...context.repo, issue_number: pr.number}); - const previous = comments.find(c => c.user.type === 'Bot' && c.body.startsWith(marker)); - if (previous) { - await github.rest.issues.updateComment({...context.repo, comment_id: previous.id, body}); - } else { - await github.rest.issues.createComment({...context.repo, issue_number: pr.number, body}); - } + with: + section: reader-api + artifact-name: pr-performance-${{ github.run_attempt }} + measurement-result: ${{ needs.compare.result }} diff --git a/.github/workflows/token-usage-performance.yml b/.github/workflows/token-usage-performance.yml new file mode 100644 index 000000000..8fc3f2439 --- /dev/null +++ b/.github/workflows/token-usage-performance.yml @@ -0,0 +1,101 @@ +name: Token usage performance +on: + pull_request: + paths: + - apps/admin-api/src/automation.py + - scripts/profile_token_usage.py + - scripts/ci/token-usage-summary.py + - tests/tooling/ci/test_token_usage_comment.py + - tests/tooling/profiling/test_token_usage_summary.py + - .github/workflows/performance-report.yml + - .github/workflows/token-usage-performance.yml + workflow_dispatch: +concurrency: + group: token-usage-performance-${{ github.ref }} + cancel-in-progress: true +permissions: + contents: read +jobs: + benchmark: + name: Native ARM64 PostgreSQL 17 + runs-on: ubuntu-24.04-arm + timeout-minutes: 30 + services: + postgres: + image: postgres:17.11-bookworm + env: + POSTGRES_USER: devfeed + POSTGRES_HOST_AUTH_METHOD: trust + POSTGRES_DB: devfeed_token_profile_test + ports: + - 127.0.0.1:5432:5432 + options: >- + --health-cmd "pg_isready -U devfeed -d devfeed_token_profile_test" + --health-interval 5s --health-timeout 5s --health-retries 12 + env: + DEVFEED_DATABASE_URL: postgresql+psycopg://devfeed@127.0.0.1:5432/devfeed_token_profile_test + DEVFEED_TEST_DATABASE_URL: postgresql+psycopg://devfeed@127.0.0.1:5432/devfeed_token_profile_test + DEVFEED_REDIS_URL: redis://127.0.0.1:6379/15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 + with: + version: "0.13.5" + enable-cache: true + cache-dependency-glob: uv.lock + cache-suffix: token-usage-profile + - name: Install locked application + run: sh scripts/ci/python-bootstrap.sh . --editable --all-packages + - name: Migrate disposable database + run: uv run --locked --no-sync --no-build devfeed db upgrade + - name: Compare production-sized and larger histories + id: measure + env: + POSTGRES_CONTAINER: ${{ job.services.postgres.id }} + run: | + test "$(uname -m)" = aarch64 + uv run --locked --no-sync --no-build python scripts/profile_token_usage.py --seed --articles 100000 300000 \ + --postgres-container "$POSTGRES_CONTAINER" + - name: Summarize before and after measurements + if: always() + env: + HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + MEASUREMENT_OUTCOME: ${{ steps.measure.outcome }} + run: | + set +e + python3 scripts/ci/token-usage-summary.py \ + --head-sha "$HEAD_SHA" --run-url "$RUN_URL" --outcome "$MEASUREMENT_OUTCOME" + summary_status=$? + set -e + if [ -f reports/token-usage/summary.md ]; then + cat reports/token-usage/summary.md >> "$GITHUB_STEP_SUMMARY" + fi + exit "$summary_status" + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: token-usage-native-arm64-${{ github.run_attempt }} + path: reports/token-usage/ + if-no-files-found: warn + retention-days: 30 + report: + name: Publish token usage results + needs: benchmark + if: >- + always() && !cancelled() && github.event_name == 'pull_request' && + github.event.pull_request.head.repo.full_name == github.repository + uses: ./.github/workflows/performance-report.yml + permissions: + contents: read + pull-requests: write + with: + section: token-usage + artifact-name: token-usage-native-arm64-${{ github.run_attempt }} + measurement-result: ${{ needs.benchmark.result }} diff --git a/apps/admin-api/README.md b/apps/admin-api/README.md index 9392e4806..411abc23c 100644 --- a/apps/admin-api/README.md +++ b/apps/admin-api/README.md @@ -64,3 +64,70 @@ profiler refuses non-local hosts and database names without the `_test` suffix. complete 30/90-day report payloads and prints timings plus activity-query buffer statistics; writer samples roll back their rows and temporary trigger changes. These local synthetic measurements do not establish production latency. + +## Completed-job token usage + +The automation report aggregates each job's cumulative `usage.totalTokens` once on its +UTC completion date. Two streaming subqueries extract token text and validate its numeric +value once per job. Literal `OFFSET 0` fences prevent PostgreSQL from pulling those +expressions into every aggregate, while avoiding a full-window materialized working set. +Projecting the date before grouping also gives PostgreSQL 17 a small daily group estimate +instead of inheriting the near-unique completion timestamp distribution. This preserves +inclusive time boundaries, succeeded/failed jobs, missing or malformed telemetry, +durations, and the eighteen-digit token validity rule. It requires no migration or +additional writes; rolling back the application restores the previous query. + +PostgreSQL maintains `OFFSET 0` as an explicit optimization fence; see the +[upstream planner change](https://www.postgresql.org/message-id/E1pK7VD-0059mE-Jc%40gemulon.postgresql.org). +Actual execution plans verify the extraction and validation projections, daily aggregation, +and temporary writes in both custom and generic plans. + +To reproduce the before/after comparison on a migrated disposable local PostgreSQL 17: + +```bash +DEVFEED_TEST_DATABASE_URL=postgresql+psycopg://.../devfeed_token_profile_test \ + uv run --locked --no-sync --no-build python scripts/profile_token_usage.py \ + --seed --articles 100000 300000 +``` + +The profiler writes `reports/token-usage/profile.json` in the checkout. The summary +renderer reads that artifact and writes `reports/token-usage/summary.md`; neither tool +accepts arbitrary file paths, and report paths must not be symlinks. + +`--seed` replaces articles, proposals and their dependent histories. Remote hosts, +connection routing overrides and database names without `_test` are refused. The tool +uses actual migrated tables and indexes, two retained article/topic jobs per article, +180 days of skewed completion history, failed/incomplete jobs, mixed telemetry and +incompressible 4 KiB provider metadata on every fifth job. The 100,000-article fixture +has approximately 60,000 article and 160,000 topic completions in a month, plus a +recent article burst. Research histories contain 73,000 and 219,000 jobs. Exact +read-only production counts at 10:05 UTC on 24 October 2026 were 92,598 articles, +156,021 article jobs, 159,575 topic jobs and 67,014 research jobs; planner and live-tuple +statistics are not exact population counts. The profiler records this comparison and +requires every 100,000+ article fixture to cover production's total histories and +qualifying 1/7/30-day counts for each job type. Smaller fixtures remain useful for +checking the full benchmark lifecycle locally. + +For 1/7/30 UTC calendar-day windows, the tool checks exact result equality, alternating +warm samples, fully parameterized custom/generic plans, aggregate group estimates, +temporary writes and runtime with a noise allowance. It retains raw `EXPLAIN (ANALYZE, +BUFFERS, SETTINGS)` plans at 4 MiB `work_mem`, JIT enabled and up to two parallel workers. +Concurrent ingestion inserts articles and starts/finishes jobs after the fixed report +cutoff, preserving comparison results; samples report actual write overlap and latency. +Reporting-thread CPU is recorded separately from elapsed time. On Linux cgroup v2, +`--postgres-container CONTAINER` also records database-container CPU, including the +concurrent writer. Use an isolated container for that measurement. + +The `Token usage performance` PR workflow runs the complete comparison on native +ARM64 PostgreSQL 17 and retains its JSON evidence and Markdown summary for 30 days. +The job summary shows before/after tables with signed improvements; same-repository +PRs also update one existing performance comment shared with the reader API comparison. +The two jobs preserve each other's sections and serialize comment updates. Positive +percentages in the token benchmark mean improvement, +negative percentages mean regression, and missing or failed evidence cannot report a +pass. The comment includes validation totals and writer-latency caveats, with resource +and production-volume details available to expand. Forks retain the artifact and job +summary without granting comment permissions. Synthetic measurements +establish the query behavior at those fixture sizes; after deployment, compare fresh +production query statistics under matched traffic and watch report freshness, database +CPU and ingestion latency before attributing an improvement to the change. diff --git a/apps/admin-api/src/automation.py b/apps/admin-api/src/automation.py index ef4dc25f7..b70571a19 100644 --- a/apps/admin-api/src/automation.py +++ b/apps/admin-api/src/automation.py @@ -112,13 +112,15 @@ def analysis_token_activity(session, start: datetime, now: datetime): ("topic_analysis", TopicAnalysisJob), ("research_verification", ResearchVerificationJob), ): - # Keep a narrow materialization boundary: otherwise PostgreSQL inlines - # JSON extraction into every SUM/COUNT/CASE and repeatedly detoasts the - # same potentially large job usage. Nineteen characters preserve the - # eighteen-digit validity rule even for oversized malformed telemetry. - usage = ( + # Two OFFSET 0 fences keep JSON extraction and numeric validation once + # per row while streaming values directly into the daily aggregates. + # Project the UTC date before grouping so its estimate does not inherit + # near-unique completion timestamps. Literal offsets also let generic + # plans remove the runtime Limit nodes. Nineteen characters preserve + # the eighteen-digit validity rule for oversized malformed telemetry. + raw_usage = ( select( - model.finished_at, + func.date(func.timezone("UTC", model.finished_at)).label("date"), model.duration_ms, func.left(model.usage["totalTokens"].astext, 19).label("raw_tokens"), ) @@ -127,12 +129,18 @@ def analysis_token_activity(session, start: datetime, now: datetime): model.finished_at <= now, model.status.in_(("succeeded", "failed")), ) - .cte(f"{kind}_usage") - .prefix_with("MATERIALIZED") + .offset(literal_column("0")) + .subquery(f"{kind}_raw_usage") + ) + raw = raw_usage.c.raw_tokens + validated = case((raw.op("~")(r"^[0-9]{1,18}$"), cast(raw, BigInteger)), else_=None) + usage = ( + select(raw_usage.c.date, raw_usage.c.duration_ms, validated.label("tokens")) + .offset(literal_column("0")) + .subquery(f"{kind}_usage") ) - day = func.date(func.timezone("UTC", usage.c.finished_at)) - raw = usage.c.raw_tokens - tokens = case((raw.op("~")(r"^[0-9]{1,18}$"), cast(raw, BigInteger)), else_=None) + day = usage.c.date + tokens = usage.c.tokens statements.append( select( day.label("date"), diff --git a/apps/extensions/tests/preview-retention-fixture.test.mjs b/apps/extensions/tests/preview-retention-fixture.test.mjs new file mode 100644 index 000000000..aa4d72c60 --- /dev/null +++ b/apps/extensions/tests/preview-retention-fixture.test.mjs @@ -0,0 +1,102 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { runInNewContext } from "node:vm"; +import { JSDOM } from "jsdom"; +import { + retentionBaselineReady, + stableRetentionPosition, +} from "../../../scripts/testing/preview-feed-retention.mjs"; + +function ready({ + count = 25, + unhydrated = -1, + placeholder = false, + fonts = "loaded", + layout = "stacked", + guest = false, + missingBookmark = false, +} = {}) { + const compact = layout === "compact"; + const rows = Array.from({ length: count }, (_, index) => { + const bookmark = missingBookmark + ? "" + : guest + ? 'Sign in to save' + : ``; + if (!compact) + return `
${index === unhydrated ? "" : '0'}${bookmark}
`; + return `${bookmark}`; + }).join(""); + const dom = new JSDOM( + `${compact ? `${rows}
` : rows}${placeholder ? '
' : ""}`, + ); + dom.window.Element.prototype.checkVisibility = () => true; + Object.defineProperty(dom.window.document, "fonts", { value: { status: fonts } }); + try { + return runInNewContext(`(${retentionBaselineReady})(".article-row")`, { + document: dom.window.document, + }); + } finally { + dom.window.close(); + } +} + +test("retention measurements wait for every paginated row's engagement hydration", () => { + assert.equal(ready(), true); + assert.equal(ready({ count: 24 }), false); + assert.equal(ready({ unhydrated: 24 }), false); +}); + +test("retention measurements wait for placeholder exit and downloaded fonts", () => { + assert.equal(ready({ placeholder: true }), false); + assert.equal(ready({ fonts: "loading" }), false); + assert.equal(ready(), true); +}); + +test("compact guest rows are ready with bookmark links and no engagement counters", () => { + assert.equal(ready({ layout: "compact", guest: true }), true); + assert.equal(ready({ layout: "compact", guest: true, missingBookmark: true }), false); +}); + +test("compact member rows wait until every bookmark button finishes hydrating", () => { + assert.equal(ready({ layout: "compact" }), true); + assert.equal(ready({ layout: "compact", unhydrated: 24 }), false); + assert.equal(ready({ layout: "compact", missingBookmark: true }), false); +}); + +for (const layout of ["stacked", "cards"]) + test(`${layout} guest bookmark links do not bypass counter hydration`, () => { + assert.equal(ready({ layout, guest: true }), true); + assert.equal(ready({ layout, guest: true, unhydrated: 24 }), false); + }); + +function stablePosition(positionAtFrame) { + let now = 0; + let frame = 0; + const callbacks = []; + const promise = runInNewContext(`(${stableRetentionPosition})(node)`, { + node: { getBoundingClientRect: () => ({ y: positionAtFrame(frame), height: 24 }) }, + document: { documentElement: { scrollHeight: 5000 } }, + scrollY: 2500, + performance: { now: () => now }, + requestAnimationFrame: (callback) => callbacks.push(callback), + }); + while (callbacks.length) { + now += 16; + frame += 1; + callbacks.shift()(); + } + return { promise, elapsed: now }; +} + +test("the baseline follows a late layout change before taking a reading-position sample", async () => { + const { promise, elapsed } = stablePosition((frame) => (frame < 7 ? 365.25 : 338.25)); + assert.deepEqual(Array.from(await promise), [338.25, 24, 2500, 5000]); + assert.ok(elapsed >= 7 * 16 + 200); +}); + +test("an unsettled baseline fails instead of hiding continued layout motion", async () => { + const { promise, elapsed } = stablePosition((frame) => frame); + await assert.rejects(promise, /Retention baseline did not settle/); + assert.ok(elapsed >= 5000); +}); diff --git a/scripts/ci/token-usage-summary.py b/scripts/ci/token-usage-summary.py new file mode 100644 index 000000000..2e0e5ccf6 --- /dev/null +++ b/scripts/ci/token-usage-summary.py @@ -0,0 +1,516 @@ +"""Render a token-report benchmark artifact as a concise, fail-closed PR comment.""" + +import argparse +import html +import json +import math +import os +import re +import stat +import statistics +from contextlib import ExitStack, suppress +from typing import Any +from urllib.parse import urlparse + +SCALES = (100000, 300000) +WINDOWS = (1, 7, 30) +MODES = ("custom", "generic") +KINDS = ("article_analysis", "topic_analysis", "research_verification") +FIELDS = ("total_rows", "one_day_rows", "seven_day_rows", "thirty_day_rows") +CHECKS = ( + "grouping_estimates_not_worse", + "grouping_estimates_bounded", + "aggregate_disk_usage_zero", + "temporary_writes_zero", + "no_sort_spills", + "buffer_accesses_not_amplified", + "runtime_within_noise_allowance", +) +MAX_PROFILE_BYTES = 5 * 1024 * 1024 + + +def mapping(value: Any) -> dict: + return value if isinstance(value, dict) else {} + + +def sequence(value: Any) -> list: + return value if isinstance(value, list) else [] + + +def number(value: Any) -> float | None: + if isinstance(value, (int, float)) and not isinstance(value, bool): + try: + if math.isfinite(value) and value >= 0: + return float(value) + except OverflowError: + pass + return None + + +def display(value: Any, *, integer: bool = False) -> str: + valid = number(value) + return "n/a" if valid is None else f"{valid:,.0f}" if integer else f"{valid:,.2f}" + + +def improvement(before: Any, after: Any) -> str: + first, second = number(before), number(after) + if first is None or second is None or first == 0: + return "n/a" + return f"{(first - second) / first * 100:+.1f}%" + + +def safe_text(value: Any) -> str: + value = html.escape(str(value), quote=True).replace("\n", " ").replace("\r", " ") + return re.sub(r"([\\`|\[\]])", r"\\\1", value)[:200] + + +def samples(strategy: dict, field: str) -> float | None: + values = [number(mapping(sample).get(field)) for sample in sequence(strategy.get("samples"))] + if not values or any(value is None for value in values): + return None + return statistics.median(value for value in values if value is not None) + + +def peak_memory(strategy: dict) -> float | None: + # Sorted aggregates do not report hash memory. Do not invent zero for them. + values = [ + number(mapping(aggregate).get("Peak Memory Usage")) + for aggregate in sequence(strategy.get("aggregates")) + if "Peak Memory Usage" in mapping(aggregate) + ] + if not values or any(value is None for value in values): + return None + return max(value for value in values if value is not None) + + +def writer_samples(strategy: dict) -> list[float] | None: + batches = sequence(strategy.get("ingestion")) + values = [ + number(value) for batch in batches for value in sequence(mapping(batch).get("latency_ms")) + ] + if not batches or not values or any(value is None for value in values): + return None + for batch in batches: + batch = mapping(batch) + count, overlapping = ( + number(batch.get("insert_and_finish")), + number(batch.get("completed_during_report")), + ) + if ( + count != len(sequence(batch.get("latency_ms"))) + or overlapping is None + or count is None + or overlapping > count + ): + return None + return [value for value in values if value is not None] + + +def p95(values: list[float] | None) -> float | None: + return sorted(values)[math.ceil(len(values) * 0.95) - 1] if values else None + + +def table(headers: tuple[str, ...], rows: list[list[str]]) -> list[str]: + return [ + "| " + " | ".join(headers) + " |", + "| " + " | ".join("---" for _ in headers) + " |", + *("| " + " | ".join(row) + " |" for row in rows), + ] + + +def build_summary( + data: Any, *, head_sha: str, run_url: str, outcome: str, error: str | None = None +) -> str: + """Missing measurements are n/a; missing cases or gates can never produce PASS.""" + missing, failures = [], [] + if error: + missing.append(error) + if not isinstance(data, dict): + missing.append("Profile root must be a JSON object.") + data = mapping(data) + scales = {} + for scale in sequence(data.get("scales")): + scale = mapping(scale) + size = number(scale.get("articles")) + if size not in SCALES or size in scales: + missing.append("Unexpected or duplicate fixture scale.") + continue + scales[size] = scale + reports = {} + for size, scale in scales.items(): + for report in sequence(scale.get("reports")): + report = mapping(report) + days, mode = number(report.get("days")), report.get("plan_mode") + report_key = (size, days, mode) if isinstance(mode, str) else None + if days not in WINDOWS or mode not in MODES or report_key in reports: + missing.append("Unexpected or duplicate report window/plan.") + continue + reports[report_key] = report + + passed_checks, equal, complete = 0, 0, 0 + runtime_rows, resource_rows, writer_rows = [], [], [] + for size in SCALES: + for days in WINDOWS: + for mode in MODES: + context = f"{size:,} articles / {days}d / {mode}" + report = reports.get((size, days, mode), {}) + strategies = mapping(report.get("strategies")) + before, after = (mapping(strategies.get(name)) for name in ("original", "current")) + first, second = number(before.get("median_ms")), number(after.get("median_ms")) + runtime_rows.append( + [ + f"{size:,}", + f"{days}d", + mode, + display(first), + display(second), + improvement(first, second), + ] + ) + if first is None or second is None: + missing.append(f"{context}: runtime measurement missing or invalid.") + else: + complete += 1 + equality = report.get("result_equality") + if equality is True: + equal += 1 + elif equality is False: + failures.append(f"{context}: token accounting differs.") + else: + missing.append(f"{context}: accounting comparison missing.") + checks = mapping(report.get("checks")) + for name in CHECKS: + if checks.get(name) is True: + passed_checks += 1 + elif checks.get(name) is False: + failures.append(f"{context}: {name}.") + else: + missing.append(f"{context}: {name} missing or invalid.") + for name, strategy in (("before", before), ("after", after)): + sample_values = sequence(strategy.get("samples")) + if any( + samples(strategy, field) is None + for field in ( + "elapsed_ms", + "application_cpu_ms", + "database_container_cpu_ms", + ) + ): + missing.append(f"{context}: {name} CPU/runtime samples missing or invalid.") + aggregates = sequence(strategy.get("aggregates")) + buffers = mapping(strategy.get("buffers")) + if ( + not aggregates + or any( + number(mapping(group).get("Plan Rows")) is None for group in aggregates + ) + or any( + "Peak Memory Usage" in mapping(group) + and number(group["Peak Memory Usage"]) is None + for group in aggregates + ) + or any( + number(buffers.get(field)) is None + for field in ("Temp Read Blocks", "Temp Written Blocks") + ) + ): + missing.append(f"{context}: {name} plan measurements missing or invalid.") + if ( + len(sequence(strategy.get("ingestion"))) != len(sample_values) + or writer_samples(strategy) is None + ): + missing.append(f"{context}: {name} writer measurements missing or invalid.") + if days == 30: + for metric, getter in ( + ( + "DB container CPU (ms)", + lambda strategy: samples(strategy, "database_container_cpu_ms"), + ), + ("App CPU (ms)", lambda strategy: samples(strategy, "application_cpu_ms")), + ( + "Temp writes (blocks)", + lambda strategy: mapping(strategy.get("buffers")).get( + "Temp Written Blocks" + ), + ), + ("Largest hash aggregate (KiB)", peak_memory), + ): + first, second = getter(before), getter(after) + resource_rows.append( + [ + f"{size:,}", + mode, + metric, + display(first), + display(second), + improvement(first, second), + ] + ) + old_writes, new_writes = writer_samples(before), writer_samples(after) + old_p95, new_p95 = p95(old_writes), p95(new_writes) + if ( + old_p95 is not None + and new_p95 is not None + and old_writes is not None + and new_writes is not None + and new_p95 > old_p95 * 1.2 + ): + writer_rows.append( + [ + f"{size:,} / {days}d / {mode}", + display(old_p95), + display(new_p95), + improvement(old_p95, new_p95), + f"{len(old_writes)} / {len(new_writes)}", + ] + ) + + baseline = mapping(data.get("production_volume")) + coverage_rows, volume_passed = [], 0 + volume_fields = [("Articles", "articles", None)] + [ + ( + f"{kind.replace('_', ' ')} / {field.replace('_rows', '').replace('_', ' ')}", + f"{kind}.{field}", + (kind, field), + ) + for kind in KINDS + for field in FIELDS + ] + for label, key, history in volume_fields: + production = ( + baseline.get("articles") + if history is None + else mapping(mapping(baseline.get("histories")).get(history[0])).get(history[1]) + ) + row = [label, display(production, integer=True)] + for size in SCALES: + scale = scales.get(size, {}) + actual = ( + scale.get("articles") + if history is None + else mapping(mapping(scale.get("histories")).get(history[0])).get(history[1]) + ) + first, second = number(production), number(actual) + recorded = mapping(scale.get("production_volume_checks")).get(key) + if first is None or second is None or not isinstance(recorded, bool): + missing.append(f"{size:,} articles: production volume evidence missing for {key}.") + status = "n/a" + elif second < first or recorded is False: + failures.append(f"{size:,} articles: production volume not covered for {key}.") + status = "FAIL" + else: + volume_passed += 1 + status = "PASS" + row.append(f"{display(actual, integer=True)} ({status})") + coverage_rows.append(row) + + if outcome == "failure": + failures.append( + "Benchmark measurement step failed; " + "partial passing measurements do not establish success." + ) + elif outcome != "success": + missing.append(f"Benchmark measurement step outcome: {safe_text(outcome)}.") + verdict = "FAIL" if failures else "INCOMPLETE" if missing else "PASS" + reason = { + "PASS": "all 12 comparisons, accounting checks and production-volume gates passed", + "FAIL": "benchmark failed; review the failed checks below", + "INCOMPLETE": "benchmark evidence is incomplete; no performance pass can be claimed", + }[verdict] + parsed_url = urlparse(run_url) + run = ( + f"[Benchmark run]({run_url})" + if parsed_url.scheme == "https" + and parsed_url.netloc == "github.com" + and re.fullmatch(r"/[A-Za-z0-9_./-]+", parsed_url.path) + and not parsed_url.query + and not parsed_url.fragment + else "Benchmark run link unavailable" + ) + architecture = data.get("client_architecture") + architecture = architecture if architecture in ("aarch64", "arm64", "x86_64") else "unknown" + lines = [ + f"**{verdict} — Token usage benchmark:** {reason}.", + "", + f"Compared head: `{safe_text(head_sha)}` · {run} · Architecture: `{architecture}`.", + "", + "Before = frozen pre-fix SQL; after = application SQL at the compared head. " + "Improvement = `(before − after) / before × 100`: **+ means better, − means worse**; " + "`n/a` means missing evidence or a zero baseline. " + "Lower is better for every performance metric.", + "", + "**Query runtime — median milliseconds**", + "", + *table( + ("Articles", "Window", "Plan", "Before (ms)", "After (ms)", "Improvement (+/−)"), + runtime_rows, + ), + "", + "**Validation**", + "", + *table( + ("Check", "Passed / expected"), + [ + ["Runtime comparisons", f"{complete} / 12"], + ["Identical token accounting", f"{equal} / 12"], + ["Runtime, plan and spill gates", f"{passed_checks} / 84"], + ["Production-volume gates", f"{volume_passed} / 26"], + ["Measurement step", safe_text(outcome)], + ], + ), + "", + "Runtime gate allows up to 20% or 20 ms noise; PASS does not mean every case got faster. " + "Writer latency is observational and has no pass/fail gate. " + "Its samples include writer startup and reports may finish " + "before a subsequent write completes.", + ] + if writer_rows: + lines.extend( + [ + "", + "**Writer latency caveats — p95 regressions over 20%**", + "", + *table( + ( + "Fixture / window / plan", + "Before p95 (ms)", + "After p95 (ms)", + "Improvement (+/−)", + "Samples before / after", + ), + writer_rows, + ), + "", + "p95 uses the nearest-rank sample. Small sample counts and startup outliers " + "prevent a sustained-load latency conclusion.", + ] + ) + for title, entries in (("Failed checks", failures), ("Missing or invalid evidence", missing)): + if entries: + lines.extend(["", f"**{title}**", "", *(f"- {entry}" for entry in entries[:20])]) + if len(entries) > 20: + lines.append( + f"- {len(entries) - 20} more; inspect the JSON artifact in the benchmark run." + ) + lines.extend( + [ + "", + "
", + "30-day CPU, temporary writes and hash memory", + "", + *table( + ("Articles", "Plan", "Metric", "Before", "After", "Improvement (+/−)"), + resource_rows, + ), + "", + "DB CPU is the container CPU median, including concurrent ingestion; " + "elapsed time is measured separately. " + "Hash memory is the largest reported aggregate node, not total process memory. " + "Temporary writes are PostgreSQL blocks from EXPLAIN.", + "", + "
", + "", + "
", + "Production-volume coverage: retained history and 1/7/30-day counts", + "", + *table( + ("Fixture", "Production snapshot", "100,000 articles", "300,000 articles"), + coverage_rows, + ), + "", + *( + [f"Production snapshot observed at: `{safe_text(baseline['observed_at'])}`.", ""] + if baseline.get("observed_at") + else [] + ), + "Production counts are the captured snapshot, not a live production benchmark.", + "", + "
", + ] + ) + return "\n".join(lines) + "\n" + + +def render(contents: str | None, *, head_sha: str, run_url: str, outcome: str) -> str: + try: + data = json.loads(contents) if contents is not None else None + if contents is None: + raise ValueError("Profile unavailable") + error = None + except ValueError: + data, error = {}, "Profile artifact is missing, unreadable or malformed." + return build_summary(data, head_sha=head_sha, run_url=run_url, outcome=outcome, error=error) + + +def read_profile(directory: int) -> str | None: + """Read only the regular profile file inside the already-open report directory.""" + try: + descriptor = os.open( + "profile.json", os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory + ) + with os.fdopen(descriptor, "rb") as profile: + metadata = os.fstat(profile.fileno()) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_PROFILE_BYTES: + return None + contents = profile.read(MAX_PROFILE_BYTES + 1) + if len(contents) > MAX_PROFILE_BYTES: + return None + return contents.decode("utf-8") + except (OSError, UnicodeError): + return None + + +def write_summary(directory: int, contents: str) -> None: + # Check before truncating: a preexisting symlink or hard link must not let + # an artifact redirect the output into a different file. + descriptor = os.open( + "summary.md", + os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, + 0o600, + dir_fd=directory, + ) + with os.fdopen(descriptor, "w", encoding="utf-8") as output: + metadata = os.fstat(output.fileno()) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: + raise ValueError("Summary output must be a regular file with one link") + os.fchmod(output.fileno(), 0o600) + output.truncate(0) + output.write(contents) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--head-sha", required=True) + parser.add_argument("--run-url", required=True) + parser.add_argument( + "--outcome", choices=("success", "failure", "skipped", "cancelled"), required=True + ) + args = parser.parse_args() + try: + # Every component and filename is fixed. Directory descriptors and + # O_NOFOLLOW keep artifact symlinks from escaping reports/token-usage. + with ExitStack() as files: + directory = os.open(".", os.O_RDONLY | os.O_DIRECTORY) + files.callback(os.close, directory) + for component in ("reports", "token-usage"): + with suppress(FileExistsError): + os.mkdir(component, dir_fd=directory) + directory = os.open( + component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory + ) + files.callback(os.close, directory) + summary = render( + read_profile(directory), + head_sha=args.head_sha, + run_url=args.run_url, + outcome=args.outcome, + ) + write_summary(directory, summary) + except (OSError, ValueError) as error: + parser.exit(2, f"Cannot safely access reports/token-usage: {error}\n") + if not summary.startswith("**PASS"): + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/scripts/profile_token_usage.py b/scripts/profile_token_usage.py new file mode 100644 index 000000000..719a7b063 --- /dev/null +++ b/scripts/profile_token_usage.py @@ -0,0 +1,614 @@ +"""Compare the original token report and current application SQL on disposable PostgreSQL. + +Requires a migrated, explicitly local DEVFEED_TEST_DATABASE_URL ending in _test. +--seed replaces articles and topic proposals, including their dependent job histories. +Never point this tool at shared development data. Raw EXPLAIN plans accompany the report. +""" + +import argparse +import hashlib +import json +import os +import platform +import stat +import statistics +import subprocess +import threading +import time +from contextlib import ExitStack, contextmanager, suppress +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from devfeed_admin_api.automation import analysis_token_activity +from devfeed_core.models import ArticleAnalysisJob, ResearchVerificationJob, TopicAnalysisJob +from sqlalchemy import BigInteger, case, cast, create_engine, func, literal, select, text, union_all +from sqlalchemy.dialects.postgresql.psycopg import PGDialect_psycopg +from sqlalchemy.engine import make_url + +NOW = datetime(2026, 10, 24, 12, tzinfo=UTC) +MODELS = ( + ("article_analysis", ArticleAnalysisJob), + ("topic_analysis", TopicAnalysisJob), + ("research_verification", ResearchVerificationJob), +) +REPORT_NAME = "reports/token-usage/profile.json" +ROOT = Path(__file__).resolve().parents[1] +PRODUCTION_VOLUME: dict[str, Any] = { + "observed_at": "2026-10-24T10:05:40Z", + "articles": 92598, + "histories": { + "article_analysis": { + "total_rows": 156021, + "one_day_rows": 3034, + "seven_day_rows": 16692, + "thirty_day_rows": 50402, + }, + "topic_analysis": { + "total_rows": 159575, + "one_day_rows": 2381, + "seven_day_rows": 30257, + "thirty_day_rows": 128657, + }, + "research_verification": { + "total_rows": 67014, + "one_day_rows": 190, + "seven_day_rows": 3142, + "thirty_day_rows": 45316, + }, + }, +} + + +def original_statement(start, end): + """Frozen pre-#295 SQL: unique completion timestamps cross the CTE boundary.""" + statements = [] + for kind, model in MODELS: + usage = ( + select( + model.finished_at, + model.duration_ms, + func.left(model.usage["totalTokens"].astext, 19).label("raw_tokens"), + ) + .where( + model.finished_at >= start, + model.finished_at <= end, + model.status.in_(("succeeded", "failed")), + ) + .cte(f"{kind}_usage") + .prefix_with("MATERIALIZED") + ) + day = func.date(func.timezone("UTC", usage.c.finished_at)) + tokens = case( + (usage.c.raw_tokens.op("~")(r"^[0-9]{1,18}$"), cast(usage.c.raw_tokens, BigInteger)), + else_=None, + ) + statements.append( + select( + day.label("date"), + literal(kind).label("kind"), + func.coalesce(func.sum(tokens), 0).label("tokens"), + func.count(tokens).label("reported"), + (func.count() - func.count(tokens)).label("unreported"), + func.coalesce(func.sum(usage.c.duration_ms), 0).label("duration"), + ).group_by(day) + ) + return union_all(*statements) + + +class Capture: + """Obtain the exact production statement without running the reporting query.""" + + def execute(self, statement): + self.statement = statement + return self + + def mappings(self): + return () + + +def current_statement(start, end): + capture = Capture() + analysis_token_activity(capture, start, end) + return capture.statement + + +def prepared_sql(statement): + # Retain every application bind, including statuses/UTC/regex/JSON keys. + # EXPLAIN EXECUTE describes actual cached generic/custom parameterized SQL. + compiled = statement.compile( + dialect=PGDialect_psycopg(paramstyle="numeric_dollar"), + compile_kwargs={"render_postcompile": True}, + ) + arguments = ",".join( + str( + literal(compiled.params[name]).compile( + dialect=PGDialect_psycopg(), compile_kwargs={"literal_binds": True} + ) + ) + for name in compiled.positiontup + ) + return str(compiled), arguments + + +def configure(connection, mode): + for setting in ( + "SET work_mem='4MB'", + "SET jit=on", + "SET max_parallel_workers_per_gather=2", + "SET track_io_timing=on", + "SET statement_timeout='90s'", + "SET timezone='Asia/Kolkata'", + f"SET plan_cache_mode='force_{mode}_plan'", + ): + connection.exec_driver_sql(setting) + connection.commit() + + +def seed(engine, articles): + """Actual migrated tables/indexes, skewed 180-day history and incompressible TOAST.""" + with engine.begin() as connection: + # A pooled measurement connection retains its bounded report timeout. + # Fixture construction has a separate budget and keeps all ordinary triggers. + connection.exec_driver_sql("SET LOCAL statement_timeout='10min'") + connection.exec_driver_sql("TRUNCATE articles,topic_proposals CASCADE") + connection.execute( + text("""INSERT INTO articles + (id,canonical_url,url_hash,title,slug,summary,content_type,feed_at,discovered_at) + SELECT md5('token-article-'||n)::uuid,'https://example.invalid/'||n, + md5('token-article-'||n),'Token profile '||n,'token-profile-'||n, + 'Disposable token reporting fixture','article',:now,:now + FROM generate_series(1,:articles) n"""), + {"now": NOW, "articles": articles}, + ) + connection.execute( + text("""INSERT INTO topic_proposals + (id,batch_id,slug,action,origin,source_name,proposed,evidence,status,created_at,created_by) + SELECT md5('token-proposal-'||n)::uuid,md5('token-profile')::uuid, + 'token-proposal-'||n,'create','import','Token profile','{}','[]','pending',:now,'{}' + FROM generate_series(1,10000) n"""), + {"now": NOW}, + ) + # Every fifth job has >4 KiB of incompressible provider/step telemetry. + # Build a bounded pool once, rather than hide seed CPU in measured samples. + connection.exec_driver_sql("""CREATE TEMP TABLE token_profile_payloads ON COMMIT DROP AS + SELECT p,string_agg(md5(p::text||'-'||n::text),'') payload + FROM generate_series(0,255) p CROSS JOIN generate_series(1,128) n GROUP BY p""") + for kind, model in MODELS[:2]: + article = kind == "article_analysis" + target = "article_id" if article else "proposal_id" + target_id = ( + "md5('token-article-'||(1+(n-1)%:articles))::uuid" + if article + else "md5('token-proposal-'||(1+(n-1)%10000))::uuid" + ) + columns = ( + "input_snapshot,catalog_snapshot,result,prompt_version" + if article + else "input_hash,input_snapshot,requested_by,result,prompt_version" + ) + values = ( + "'{}','{}','{}','profile'" if article else "repeat('a',64),'{}','{}','{}','profile'" + ) + recent = 30 if article else 80 + burst = ( + "WHEN n%100<3 THEN ((n*7919::bigint)%43200)*interval '1 second'" if article else "" + ) + connection.execute( + text(f"""INSERT INTO {model.__tablename__} + (id,{target},status,attempts,available_at,created_at,finished_at,duration_ms,usage, + {columns}) + SELECT md5('{kind}-'||n)::uuid,{target_id}, + CASE WHEN n%13=0 THEN 'failed' ELSE 'succeeded' END,1,:now,:now, + :now-CASE {burst} WHEN n%100<{recent} + THEN ((n*7919::bigint)%2592000)*interval '1 second' + ELSE interval '30 days' + +((n*15485863::bigint)%12960000)*interval '1 second' END, + n%60000, + (CASE n%11 + WHEN 0 THEN '{{}}'::jsonb + WHEN 1 THEN '{{"totalTokens": null}}'::jsonb + WHEN 2 THEN '{{"totalTokens":"broken"}}'::jsonb + WHEN 3 THEN '{{"totalTokens":-1}}'::jsonb + WHEN 4 THEN '{{"totalTokens": 1.5}}'::jsonb + WHEN 5 THEN '{{"totalTokens":"1234567890123456789"}}'::jsonb + WHEN 6 THEN '{{"totalTokens":"999999999999999999"}}'::jsonb + WHEN 7 THEN '{{"totalTokens": 0}}'::jsonb + WHEN 8 THEN jsonb_build_object('totalTokens',(n%100000)::text) + ELSE jsonb_build_object('totalTokens',n%100000) END) + || CASE WHEN n%5=0 THEN jsonb_build_object('provider_metadata',p.payload) + ELSE '{{}}'::jsonb END,{values} + FROM generate_series(1,:jobs) n + JOIN token_profile_payloads p ON p.p=n%256"""), + {"now": NOW, "articles": articles, "jobs": 2 * articles}, + ) + connection.execute( + text("""INSERT INTO research_verification_jobs + (id,status,attempts,available_at,created_at,finished_at,relationships,usage,duration_ms) + SELECT id,status,attempts,available_at,created_at,finished_at,false,usage,duration_ms + FROM topic_analysis_jobs WHERE id IN ( + SELECT md5('topic_analysis-'||n)::uuid FROM generate_series(1,:research) n) + """), + {"research": max(1, articles * 73 // 100)}, + ) + connection.execute( + text("""INSERT INTO article_analysis_jobs + (id,article_id,status,attempts,available_at,created_at,input_snapshot,catalog_snapshot,result) + SELECT md5('unfinished-'||n)::uuid,md5('token-article-'||n)::uuid, + CASE WHEN n%2=0 THEN 'running' ELSE 'queued' END,0,:now,:now,'{}','{}','{}' + FROM generate_series(1,:articles) n WHERE n%97=0"""), + {"now": NOW, "articles": articles}, + ) + # Commit before ANALYZE so every profiling connection sees identical statistics. + with engine.begin() as connection: + for table in ("articles", "topic_proposals", *(m.__tablename__ for _, m in MODELS)): + connection.exec_driver_sql(f"ANALYZE {table}") + + +def relation_sizes(connection): + return [ + dict(row) + for row in connection.execute( + text("""SELECT c.relname,c.reltuples::bigint estimated_rows, + pg_relation_size(c.oid) heap_bytes,pg_total_relation_size(c.oid) total_bytes, + CASE WHEN c.reltoastrelid=0 THEN 0 + ELSE pg_total_relation_size(c.reltoastrelid) END toast_bytes + FROM pg_class c WHERE c.oid IN ('articles'::regclass, + 'article_analysis_jobs'::regclass,'topic_analysis_jobs'::regclass, + 'research_verification_jobs'::regclass) ORDER BY c.relname""") + ).mappings() + ] + + +def history_counts(connection): + histories = {} + for kind, model in MODELS: + row = ( + connection.execute( + text(f"""SELECT count(*) total_rows, + min(finished_at) oldest_completion,max(finished_at) newest_completion, + count(DISTINCT finished_at) distinct_completions, + count(*) FILTER (WHERE finished_at>=:now-interval '12 hours') one_day_rows, + count(*) FILTER (WHERE finished_at>=:now-interval '6 days 12 hours') seven_day_rows, + count(*) FILTER (WHERE finished_at>=:now-interval '29 days 12 hours') thirty_day_rows + FROM {model.__tablename__} WHERE status IN ('succeeded','failed') + AND finished_at<=:now"""), + {"now": NOW}, + ) + .mappings() + .one() + ) + histories[kind] = dict(row) + return histories + + +def postgres_cpu(container): + """Cgroup CPU is actual database container CPU, not SQL elapsed time.""" + if not container: + return None + pid = subprocess.check_output( + ["docker", "inspect", "--format", "{{.State.Pid}}", container], text=True + ).strip() + for entry in Path(f"/proc/{pid}/cgroup").read_text().splitlines(): + if entry.startswith("0::"): + path = Path("/sys/fs/cgroup") / entry.split("::", 1)[1].lstrip("/") / "cpu.stat" + return int(dict(line.split() for line in path.read_text().splitlines())["usage_usec"]) + raise RuntimeError("Database CPU measurement requires cgroup v2") + + +class Ingestion: + def __init__(self, engine, rate): + self.engine, self.rate = engine, rate + self.stop = threading.Event() + self.ready = threading.Event() + self.count, self.errors, self.samples = 0, [], [] + self.thread = threading.Thread(target=self.run, daemon=True) + + def run(self): + try: + with self.engine.connect() as connection: + while not self.stop.is_set(): + started = time.perf_counter() + identifier = str(time.time_ns()) + connection.execute( + text("""WITH article AS ( + INSERT INTO articles + (id,canonical_url,url_hash,title,slug,summary,content_type,feed_at,discovered_at) + VALUES (gen_random_uuid(),'https://example.invalid/ingest/'||:n, + md5('ingest-'||:n),'Concurrent ingestion','ingest-'||:n, + '','article',:now,:now) + RETURNING id) + INSERT INTO article_analysis_jobs + (id,article_id,status,attempts,available_at,created_at,input_snapshot, + catalog_snapshot,result,usage) + SELECT md5('ingest-'||:n)::uuid,id,'running',1,:now,:now,'{}','{}','{}','{}' + FROM article"""), + {"n": identifier, "now": NOW + timedelta(days=1)}, + ) + connection.execute( + text("""UPDATE article_analysis_jobs + SET status='succeeded',finished_at=:now, + duration_ms=100,usage='{"totalTokens": 1000}' + WHERE id=md5('ingest-'||:n)::uuid"""), + {"n": identifier, "now": NOW + timedelta(days=1)}, + ) + connection.commit() + self.count += 1 + self.samples.append((time.perf_counter() - started) * 1000) + self.ready.set() + self.stop.wait(max(0, 1 / self.rate - (time.perf_counter() - started))) + except Exception as error: + self.errors.append(str(error)) + self.ready.set() + + def __enter__(self): + self.thread.start() + assert self.ready.wait(15), "Concurrent ingestion did not start" + assert not self.errors, self.errors + return self + + def __exit__(self, *_): + self.stop.set() + self.thread.join(timeout=15) + assert not self.thread.is_alive(), "Concurrent ingestion did not stop" + assert not self.errors, self.errors + + +def nodes(plan): + yield plan + for child in plan.get("Plans", ()): + yield from nodes(child) + + +def measure(engine, days, mode, repeats, container, rate): + start = NOW.replace(hour=0) - timedelta(days=days - 1) + results, plans = {}, {} + samples: dict[str, list[dict[str, Any]]] = {} + ingestion: dict[str, list[dict[str, Any]]] = {} + with engine.connect() as connection: + configure(connection, mode) + statements = {} + for name, builder in (("original", original_statement), ("current", current_statement)): + query, arguments = prepared_sql(builder(start, NOW)) + connection.exec_driver_sql(f"PREPARE token_{name} AS {query}") + statements[name] = f"EXECUTE token_{name}({arguments})" + for repeat in range(repeats + 1): + for name in ("original", "current") if repeat % 2 == 0 else ("current", "original"): + execute = statements[name] + # A separate writer for each strategy gives matched requested load. + with Ingestion(engine, rate) as writer: + db_cpu = postgres_cpu(container) + writes_before = writer.count + app_cpu, started = time.thread_time(), time.perf_counter() + rows = sorted(tuple(row) for row in connection.exec_driver_sql(execute)) + elapsed = (time.perf_counter() - started) * 1000 + cpu = (time.thread_time() - app_cpu) * 1000 + db_after = postgres_cpu(container) + overlapping_writes = writer.count - writes_before + results[name] = rows + if len(results) == 2: + assert results["original"] == results["current"], "Token accounting changed" + if repeat: # First pass warms both paths without dropping shared caches. + samples.setdefault(name, []).append( + { + "elapsed_ms": elapsed, + "application_cpu_ms": cpu, + "database_container_cpu_ms": None + if db_cpu is None + else (db_after - db_cpu) / 1000, + } + ) + ingestion.setdefault(name, []).append( + { + "insert_and_finish": writer.count, + "completed_during_report": overlapping_writes, + "latency_ms": writer.samples, + } + ) + if repeat == repeats: + plans[name] = connection.exec_driver_sql( + "EXPLAIN (ANALYZE,BUFFERS,VERBOSE,SETTINGS,FORMAT JSON) " + execute + ).scalar_one()[0] + connection.exec_driver_sql("DEALLOCATE ALL") + connection.commit() + summaries = {} + for name in ("original", "current"): + plan = plans[name] + groups = [node for node in nodes(plan["Plan"]) if node["Node Type"] == "Aggregate"] + summaries[name] = { + "samples": samples[name], + "median_ms": statistics.median(sample["elapsed_ms"] for sample in samples[name]), + "ingestion": ingestion[name], + "execution_ms": plan["Execution Time"], + "buffers": {key: value for key, value in plan["Plan"].items() if "Blocks" in key}, + "aggregates": [ + { + key: node[key] + for key in ( + "Node Type", + "Strategy", + "Plan Rows", + "Actual Rows", + "Peak Memory Usage", + "Disk Usage", + "HashAgg Batches", + ) + if key in node + } + for node in groups + ], + "plan": plan, + } + original_groups = sum(g["Plan Rows"] for g in summaries["original"]["aggregates"]) + current_groups = sum(g["Plan Rows"] for g in summaries["current"]["aggregates"]) + accesses = { + name: sum( + summary["buffers"].get(key, 0) for key in ("Shared Hit Blocks", "Shared Read Blocks") + ) + for name, summary in summaries.items() + } + checks = { + # Each projected date may use PostgreSQL's 200-distinct fallback even + # for tiny smoke fixtures with fewer input rows across the three kinds. + "grouping_estimates_not_worse": current_groups <= max(original_groups, 600), + "grouping_estimates_bounded": all( + g["Plan Rows"] <= 200 for g in summaries["current"]["aggregates"] + ), + "aggregate_disk_usage_zero": all( + g.get("Disk Usage", 0) == 0 for g in summaries["current"]["aggregates"] + ), + "temporary_writes_zero": summaries["current"]["buffers"].get("Temp Written Blocks", 0) == 0, + "no_sort_spills": all( + node.get("Sort Space Type") != "Disk" for node in nodes(plans["current"]["Plan"]) + ), + "buffer_accesses_not_amplified": accesses["current"] <= 1.05 * accesses["original"] + 16, + "runtime_within_noise_allowance": summaries["current"]["median_ms"] + <= max(1.2 * summaries["original"]["median_ms"], summaries["original"]["median_ms"] + 20), + } + canonical = json.dumps(results["current"], default=str, separators=(",", ":")) + return { + "days": days, + "plan_mode": mode, + "result_equality": True, + "result_sha256": hashlib.sha256(canonical.encode()).hexdigest(), + "returned_groups": len(results["current"]), + "checks": checks, + "strategies": summaries, + } + + +@contextmanager +def report_directory(): + """Open checkout-owned directories without following links, including rename races.""" + with ExitStack() as descriptors: + directory = os.open(ROOT, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + descriptors.callback(os.close, directory) + for component in ("reports", "token-usage"): + with suppress(FileExistsError): + os.mkdir(component, mode=0o700, dir_fd=directory) + directory = os.open( + component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory + ) + descriptors.callback(os.close, directory) + yield directory + + +def write_report(output): + with report_directory() as directory: + descriptor = os.open( + "profile.json", + os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, + 0o600, + dir_fd=directory, + ) + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + info = os.fstat(stream.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: + raise ValueError("Report output must be a regular file with one link") + os.fchmod(stream.fileno(), 0o600) + stream.truncate(0) + stream.write(json.dumps(output, indent=2, default=str) + "\n") + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--seed", action="store_true") + parser.add_argument("--articles", nargs="+", type=int, default=[100000, 300000]) + parser.add_argument("--repeats", type=int, default=3) + parser.add_argument("--ingestion-rate", type=int, default=20) + parser.add_argument("--postgres-container") + args = parser.parse_args(argv) + url = make_url(os.environ["DEVFEED_TEST_DATABASE_URL"]) + if ( + url.get_backend_name() != "postgresql" + or url.host not in {"127.0.0.1", "localhost"} + or not (url.database or "").endswith("_test") + or url.query + or any(os.environ.get(key) for key in ("PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE")) + ): + parser.error("Use an explicitly disposable local database ending in _test") + if not ( + all(100 <= count <= 1000000 for count in args.articles) + and 1 <= args.repeats <= 10 + and 1 <= args.ingestion_rate <= 100 + ): + parser.error("Fixture or sampling sizes exceed supported profiling bounds") + if len(args.articles) > 1 and not args.seed: + parser.error("Multiple fixture sizes require explicit --seed") + engine = create_engine(url, connect_args={"prepare_threshold": None}) + output: dict[str, Any] = { + "client_architecture": platform.machine(), + "production_volume": PRODUCTION_VOLUME, + "scales": [], + } + try: + write_report(output) + with engine.connect() as connection: + assert ( + connection.exec_driver_sql("SELECT current_database()").scalar_one() == url.database + ) + output["server_version"] = connection.exec_driver_sql("SELECT version()").scalar_one() + assert "PostgreSQL 17." in output["server_version"] + for count in args.articles: + if args.seed: + print(f"Seeding {count} articles and retained job histories", flush=True) + seed(engine, count) + with engine.connect() as connection: + scale = { + "articles": count, + "relations": relation_sizes(connection), + "histories": history_counts(connection), + "reports": [], + } + output["scales"].append(scale) + scale["production_volume_checks"] = { + "articles": count >= PRODUCTION_VOLUME["articles"], + **{ + f"{kind}.{field}": scale["histories"][kind][field] >= minimum + for kind, fields in PRODUCTION_VOLUME["histories"].items() + for field, minimum in fields.items() + }, + } + for mode in ("custom", "generic"): + for days in (1, 7, 30): + scale["reports"].append( + measure( + engine, + days, + mode, + args.repeats, + args.postgres_container, + args.ingestion_rate, + ) + ) + write_report(output) + print( + f"{count} articles / {days} days / {mode}: " + + " / ".join( + f"{name} {value['median_ms']:.1f} ms" + for name, value in scale["reports"][-1]["strategies"].items() + ), + flush=True, + ) + finally: + engine.dispose() + failures = [ + f"{scale['articles']} articles/{report['days']} days/{report['plan_mode']}: {name}" + for scale in output["scales"] + for report in scale["reports"] + for name, passed in report["checks"].items() + if not passed + ] + failures.extend( + f"{scale['articles']} articles: fixture below production {name}" + for scale in output["scales"] + if scale["articles"] >= 100000 + for name, passed in scale["production_volume_checks"].items() + if not passed + ) + if failures: + raise SystemExit("Benchmark checks failed (full evidence saved):\n" + "\n".join(failures)) + + +if __name__ == "__main__": + main() diff --git a/scripts/testing/preview-feed-retention.mjs b/scripts/testing/preview-feed-retention.mjs index 16f23d6bf..6fe05c2b7 100644 --- a/scripts/testing/preview-feed-retention.mjs +++ b/scripts/testing/preview-feed-retention.mjs @@ -71,6 +71,47 @@ function requestCounts(requests) { return counts; } +export function retentionBaselineReady(selector) { + const rows = [...document.querySelectorAll(selector)].filter((row) => row.checkVisibility()); + return ( + rows.length === 25 && + rows.every((row) => { + // Compact rows render bookmarks without the like/open counters used by + // Stacked and Grid. Guest bookmarks are links; member buttons enable only + // after their engagement record has loaded. + if (!row.matches(".article-list-row")) return row.querySelector(".engagement-count"); + const bookmark = row.querySelector(".bookmark-button"); + return bookmark?.matches("a[href]") || (bookmark?.matches("button") && !bookmark.disabled); + }) && + !document.querySelector(".reader-loading-placeholder") && + document.fonts.status === "loaded" + ); +} + +/** Capture the baseline after hydration and native scrolling have stopped moving it. */ +export function stableRetentionPosition(node) { + return new Promise((resolve, reject) => { + const started = performance.now(); + let stableSince = started; + let previous; + const measure = () => { + const bounds = node.getBoundingClientRect(); + const position = [bounds.y, bounds.height, scrollY, document.documentElement.scrollHeight]; + const key = JSON.stringify(position); + const now = performance.now(); + if (key !== previous) { + previous = key; + stableSince = now; + } + if (now - stableSince >= 200) resolve(position); + else if (now - started >= 5000) + reject(new Error(`Retention baseline did not settle: ${key}`)); + else requestAnimationFrame(measure); + }; + requestAnimationFrame(measure); + }); +} + async function previewDuringNativeScroll(page, title, target) { await page.bringToFront(); // Activate the actual link in the first moving browser frame. A driver roundtrip @@ -185,11 +226,15 @@ export async function checkPreviewFeedRetention( // Keep the trigger clear of the sticky header before recording scroll; Playwright otherwise // legitimately scrolls an obscured Stacked title while preparing its click. await title.evaluate((node) => node.scrollIntoView({ block: "center", behavior: "instant" })); - // Wait for engagement hydration and the initial placeholder's exit transition. - await page.waitForTimeout(250); + // A fixed delay can record the baseline while a late engagement response or + // placeholder exit is still changing row heights on a busy browser runner. + await page.waitForFunction(retentionBaselineReady, rowSelector); + await title.evaluate(stableRetentionPosition); if (resumeOnly) { const resumedRows = await rows.elementHandles(); const resumeBounds = await title.boundingBox(); + const beforeScroll = await page.evaluate(() => scrollY); + const resumeRequestStart = requests().length; const resumeTime = await page.evaluate(() => Date.now() + 120_000); await page.clock.setFixedTime(resumeTime); const resumedAuth = page.waitForResponse((response) => @@ -201,27 +246,38 @@ export async function checkPreviewFeedRetention( }); await resumedAuth; await page.waitForTimeout(300); - assert.equal(await rows.count(), 25, "Tab resume retains every paginated article"); + await page.waitForFunction(retentionBaselineReady, rowSelector); + const rowCount = await rows.count(); + const connected = []; for (const row of resumedRows) { - assert.equal( - await row.evaluate((node) => node.isConnected), - true, - "Tab resume retains the original article DOM", - ); + connected.push(await row.evaluate((node) => node.isConnected)); await row.dispose(); } + const resumedBounds = await title.boundingBox(); + results.push({ + view, + retained: connected.filter(Boolean).length, + rowCount, + readingPositionDelta: resumedBounds ? resumedBounds.y - resumeBounds.y : null, + before: { bounds: resumeBounds, scroll: beforeScroll }, + after: { bounds: resumedBounds, scroll: await page.evaluate(() => scrollY) }, + requests: requestCounts(requests().slice(resumeRequestStart)), + }); + // Preserve the failing scenario too; assertions previously left no artifact + // for the layout that moved, making a race indistinguishable from a regression. + if (output) { + await mkdir(dirname(output), { recursive: true }); + await writeFile(output, JSON.stringify(results, null, 2)); + } + assert.equal(rowCount, 25, "Tab resume retains every paginated article"); + assert.ok(connected.every(Boolean), "Tab resume retains the original article DOM"); + assert.ok(resumedBounds, "Tab resume retains a visible article title"); // Engagement controls can change line height during hydration. Retain the // same article within one text line rather than requiring an identical pixel. - const resumedBounds = await title.boundingBox(); assert.ok( Math.abs(resumedBounds.y - resumeBounds.y) <= 24, `Tab resume retains the reading position (${target}, ${view}, before=${resumeBounds.y}, after=${resumedBounds.y})`, ); - results.push({ - view, - retained: resumedRows.length, - readingPositionDelta: resumedBounds.y - resumeBounds.y, - }); continue; } const requestStart = requests().length; diff --git a/tests/admin/operations/test_automation_policy_unit.py b/tests/admin/operations/test_automation_policy_unit.py index 644f181e4..7b699b981 100644 --- a/tests/admin/operations/test_automation_policy_unit.py +++ b/tests/admin/operations/test_automation_policy_unit.py @@ -233,7 +233,7 @@ def test_completed_token_series_keeps_idle_days_and_combines_all_three_job_kinds assert points[2].reported_runs == 2 and duration == 275 -def test_completed_token_query_reads_each_usage_once_into_bounded_materialized_values(): +def test_completed_token_query_streams_bounded_usage_with_once_per_row_validation(): from sqlalchemy.dialects import postgresql statements = [] @@ -246,14 +246,29 @@ def execute(statement): SimpleNamespace(execute=execute), NOW - timedelta(days=7), NOW ) assert len(statements) == 1 + for branch in statements[0].selects: + usage = branch.get_final_froms()[0] + raw_usage = usage.element.get_final_froms()[0] + # Keep the computed date visible across both projection boundaries; + # grouping a timestamp expression loses its daily cardinality estimate. + projected_day = branch.selected_columns["date"].element + assert getattr(projected_day, "table", None) is usage + assert getattr(usage.element.selected_columns["date"], "table", None) is raw_usage + assert len(usage.c) == len(raw_usage.c) == 3 + assert "finished_at" not in usage.c and "finished_at" not in raw_usage.c + assert "raw_tokens" not in usage.c sql = str(statements[0].compile(dialect=postgresql.dialect())) - assert sql.count("AS MATERIALIZED") == 3 + # Literal OFFSET 0 fences preserve both streaming projections in generic + # plans, without introducing a materialized working set or runtime limits. + assert sql.count("OFFSET 0") == 6 + assert "MATERIALIZED" not in sql for table in ("article_analysis_jobs", "topic_analysis_jobs", "research_verification_jobs"): assert sql.count(f"{table}.usage ->>") == 1 assert f"{table}.finished_at >=" in sql assert f"{table}.finished_at <=" in sql assert f"{table}.status IN" in sql assert sql.count("left(") == 3 + assert sql.count("CASE WHEN") == 3 @pytest.mark.parametrize( diff --git a/tests/admin/reporting/test_overview.py b/tests/admin/reporting/test_overview.py index edfd6e13a..acf28b81d 100644 --- a/tests/admin/reporting/test_overview.py +++ b/tests/admin/reporting/test_overview.py @@ -155,19 +155,138 @@ def test_token_activity_counts_all_recorded_job_types_on_utc_finish_day(database assert data.token_activity[-1].unreported_runs == 2 +@pytest.mark.parametrize("timezone", ["Pacific/Auckland", "Pacific/Honolulu"]) +def test_token_activity_keeps_exact_completion_boundaries_statuses_and_durations( + database, timezone +): + start = START + timedelta(hours=12) + midnight = START + timedelta(days=1) + end = START + timedelta(days=3) + # The middle idle day and midnight endpoint must survive UTC bucketing even + # when the database session's local date differs from the completion date. + cases = [ + ("succeeded", start - timedelta(microseconds=1), {"totalTokens": 999}, 1000), + ("succeeded", start, {"totalTokens": 10}, 1), + ("failed", midnight - timedelta(microseconds=1), {"totalTokens": 20}, 2), + ("succeeded", midnight, {"totalTokens": 30}, 3), + ("failed", midnight, {"attempts": {"1": {"totalTokens": 999}}}, 4), + ("succeeded", end, {"totalTokens": 0}, 5), + ("failed", end, {"totalTokens": "invalid"}, 6), + ("succeeded", end + timedelta(microseconds=1), {"totalTokens": 999}, 1000), + ("queued", midnight, {"totalTokens": 999}, 1000), + ("running", midnight, {"totalTokens": 999}, 1000), + ("succeeded", None, {"totalTokens": 999}, 1000), + ("failed", None, {"totalTokens": 999}, 1000), + ] + with database.begin() as session: + for kind, scale in (("article", 1), ("topic", 10), ("research", 100)): + for status, finished, usage, duration in cases: + shared = dict( + status=status, + created_at=start - timedelta(days=30), + finished_at=finished, + usage=usage, + duration_ms=duration * scale, + ) + # Independent targets allow deliberately inconsistent queued and + # running completion timestamps without violating active-job keys. + if kind == "article": + content = article() + session.add(content) + session.flush() + session.add(ArticleAnalysisJob(article_id=content.id, **shared)) + else: + draft = proposal() + session.add(draft) + session.flush() + if kind == "topic": + session.add(topic_job(proposal_id=draft.id, **shared)) + else: + parent = topic_job(proposal_id=draft.id, status="succeeded") + session.add(parent) + session.flush() + session.add( + ResearchVerificationJob(id=parent.id, relationships=False, **shared) + ) + with database() as session: + session.execute( + text("SELECT set_config('TimeZone', :timezone, true)"), {"timezone": timezone} + ) + points, duration = automation.analysis_token_activity(session, start, end) + instant, instant_duration = automation.analysis_token_activity(session, end, end) + empty, empty_duration = automation.analysis_token_activity( + session, end + timedelta(days=1), end + timedelta(days=2) + ) + + assert [point.date for point in points] == [ + "2026-09-03", + "2026-09-04", + "2026-09-05", + "2026-09-06", + ] + for point in points[:2]: + assert point.article_analysis == point.topic_analysis == point.research_verification == 30 + assert [(point.reported_runs, point.unreported_runs) for point in points] == [ + (6, 0), + (3, 3), + (0, 0), + (3, 3), + ] + assert all( + point.article_analysis == point.topic_analysis == point.research_verification == 0 + for point in points[2:] + ) + assert duration == 21 * (1 + 10 + 100) + assert [point.model_dump() for point in instant] == [points[-1].model_dump()] + assert instant_duration == 11 * (1 + 10 + 100) + assert [point.date for point in empty] == ["2026-09-07", "2026-09-08"] + assert all( + point.article_analysis + == point.topic_analysis + == point.research_verification + == point.reported_runs + == point.unreported_runs + == 0 + for point in empty + ) + assert empty_duration == 0 + + @pytest.mark.parametrize("kind", ["article", "topic", "research"]) def test_token_activity_bounds_malformed_usage_without_changing_counts_or_duration(database, kind): - values = [0, 12, "0007", "9" * 18, "9" * 19, "9" * 20000, -1, 1.5, True, None, {}, []] + values = [ + 0, + 12, + "0007", + *(["9" * 18] * 10), + "0" * 18, + "9" * 19, + "9" * 20000, + -1, + 1.5, + True, + None, + {}, + [], + "", + " 12", + "12 ", + "+12", + "1e3", + "１２", + ] + usages = [{"totalTokens": value, "irrelevant": "body" * 8000} for value in values] + usages.extend([{}, {"attempts": {"1": {"totalTokens": 999}}}, None, [], "17", 17]) with database.begin() as session: content, draft = article(), proposal() session.add_all([content, draft]) session.flush() - for index, value in enumerate(values): + for index, usage in enumerate(usages): shared = dict( status="failed" if index % 2 else "succeeded", finished_at=START, duration_ms=index + 1, - usage={"totalTokens": value, "irrelevant": "body" * 8000}, + usage=usage, ) if kind == "article": session.add(ArticleAnalysisJob(article_id=content.id, **shared)) @@ -189,10 +308,10 @@ def test_token_activity_bounds_malformed_usage_without_changing_counts_or_durati "research": "research_verification", }[kind], ) - == 10**18 - 1 + 19 + == 10 * (10**18 - 1) + 19 ) - assert points[0].reported_runs == 4 and points[0].unreported_runs == len(values) - 4 - assert duration == sum(range(1, len(values) + 1)) + assert points[0].reported_runs == 14 and points[0].unreported_runs == len(usages) - 14 + assert duration == sum(range(1, len(usages) + 1)) assert all(point.reported_runs == point.unreported_runs == 0 for point in points[1:]) diff --git a/tests/tooling/ci/test_token_usage_comment.py b/tests/tooling/ci/test_token_usage_comment.py new file mode 100644 index 000000000..5da2ba9cc --- /dev/null +++ b/tests/tooling/ci/test_token_usage_comment.py @@ -0,0 +1,369 @@ +"""Exercise the shared performance publisher against mocked GitHub and artifacts.""" + +import json +import subprocess + +import pytest +import yaml + +from tests.support.paths import REPO_ROOT + +MARKER = "" +TOKEN_MARKER = "" +SHA = "a" * 40 +SUMMARY = ( + "**PASS — Token usage benchmark:** all checks passed.\n\n" + "| Window | Before (ms) | After (ms) | Improvement (+/−) |\n" + "| 30d | 100.00 | 75.00 | +25.0% |\n" + "| 1d | 10.00 | 11.00 | -10.0% |\n" +) +API_SUMMARY = "# PR reader API performance\n\n**PASS** — all latency checks passed." + + +def comment(identifier=12, *, run=122, attempt=1, login="github-actions[bot]", kind="Bot"): + return { + "id": identifier, + "user": {"login": login, "type": kind}, + "body": ( + f"{TOKEN_MARKER}\n\n" + f"Compared head: `{SHA}`\nOld token results" + ), + } + + +def publish(**settings): + workflow = yaml.safe_load((REPO_ROOT / ".github/workflows/performance-report.yml").read_text()) + source = next( + step["with"]["script"] + for step in workflow["jobs"]["publish"]["steps"] + if step.get("uses", "").startswith("actions/github-script@") + ) + harness = r""" +const assert = require('node:assert/strict'); +const input = JSON.parse(process.argv[1]); +const result = {created: [], updated: [], deleted: [], failures: [], reads: 0, pullChecks: 0}; +const sha = 'a'.repeat(40); +const context = { + payload: {pull_request: {number: 301, head: {sha}}}, + repo: {owner: 'abhi1693', repo: 'devfeed.tech'}, + serverUrl: 'https://github.com', runId: input.run ?? 123, +}; +globalThis.artifactExecuted = false; +const fs = { + lstatSync(file) { + assert.equal(file, 'report/summary.md'); + if (input.missing) throw new Error('Private error details'); + return {isFile: () => !input.symlink, size: input.size ?? Buffer.byteLength(input.summary)}; + }, + readFileSync(file, encoding) { + assert.equal(file, 'report/summary.md'); assert.equal(encoding, 'utf8'); + result.reads += 1; + return input.summary; + }, +}; +const core = { + info: () => {}, + setFailed: message => result.failures.push(message), +}; +const listComments = () => {throw new Error('Pagination must be mocked');}; +const github = { + rest: { + pulls: {get: async request => { + assert.equal(request.pull_number, 301); + result.pullChecks += 1; + const stale = input.stale || (input.headChanged && result.pullChecks > 1); + return {data: {head: {sha: stale ? 'b'.repeat(40) : sha}}}; + }}, + issues: { + listComments, + createComment: async request => result.created.push(request), + updateComment: async request => { + if (input.updateFails) throw new Error('GitHub update failed'); + result.updated.push(request); + }, + deleteComment: async request => { + assert.equal(result.updated.length, 1, 'Migrate sections before removing duplicates'); + result.deleted.push(request); + }, + }, + }, + paginate: async (method, request) => { + assert.equal(method, listComments); assert.equal(request.issue_number, 301); + return input.comments ?? []; + }, +}; +const AsyncFunction = Object.getPrototypeOf(async function() {}).constructor; +new AsyncFunction('require', 'context', 'github', 'core', 'process', input.source)( + name => {assert.equal(name, 'fs'); return fs;}, context, github, core, + {env: { + REPORT_SECTION: input.section, MEASUREMENT_RESULT: input.status, + DOWNLOAD_OUTCOME: input.download, GITHUB_RUN_ATTEMPT: input.attempt ?? '2', + }}, +).then(() => { + result.artifactExecuted = globalThis.artifactExecuted; + process.stdout.write(JSON.stringify(result)); +}).catch(error => { + result.error = error.message; + process.stdout.write(JSON.stringify(result)); +}); +""" + completed = subprocess.run( + [ + "node", + "-e", + harness, + json.dumps( + { + "source": source, + "summary": SUMMARY, + "section": "token-usage", + "status": "success", + "download": "success", + **settings, + } + ), + ], + capture_output=True, + text=True, + check=True, + timeout=10, + ) + return json.loads(completed.stdout) + + +def stored_comment(request, identifier=12): + return { + "id": identifier, + "user": {"login": "github-actions[bot]", "type": "Bot"}, + "body": request["body"], + } + + +def test_publisher_creates_one_comment_with_signed_results_and_run_link(): + result = publish() + assert not result["failures"] + assert not result["updated"] + assert len(result["created"]) == 1 + request = result["created"][0] + assert request["issue_number"] == 301 + assert request["owner"] == "abhi1693" + assert request["repo"] == "devfeed.tech" + assert request["body"].startswith(MARKER) + assert SUMMARY in request["body"] + assert "https://github.com/abhi1693/devfeed.tech/actions/runs/123" in request["body"] + assert f"head:{SHA} run:123 attempt:2" in request["body"] + assert "## Reader API performance\n\n**Pending:**" in request["body"] + assert not result["artifactExecuted"] + + +def test_publisher_adopts_oldest_actions_comment_and_removes_only_performance_duplicates(): + result = publish( + comments=[ + comment(30, login="octocat", kind="User"), + comment(29, login="renovate[bot]"), + comment(12), + comment(11), + {**comment(10), "body": "\nCoverage results"}, + ] + ) + assert not result["created"] + assert len(result["updated"]) == 1 + assert result["updated"][0]["comment_id"] == 11 + assert SUMMARY in result["updated"][0]["body"] + assert [item["comment_id"] for item in result["deleted"]] == [12] + + +@pytest.mark.parametrize( + "settings", + [ + {"login": "octocat", "kind": "User"}, + {"login": "github-actions[bot]", "kind": "User"}, + {"login": "another[bot]", "kind": "Bot"}, + ], +) +def test_publisher_does_not_overwrite_human_or_other_bot_comments(settings): + result = publish(comments=[comment(**settings)]) + assert len(result["created"]) == 1 + assert not result["updated"] + assert not result["deleted"] + + +@pytest.mark.parametrize("settings", [{"stale": True}, {"headChanged": True}]) +def test_publisher_does_not_publish_results_after_pr_head_changes(settings): + result = publish(**settings) + assert not result["created"] + assert not result["updated"] + assert not result["deleted"] + assert not result["failures"] + if settings.get("stale"): + assert result["reads"] == 0 + else: + assert result["pullChecks"] == 2 + + +@pytest.mark.parametrize( + "previous", [comment(run=124), comment(run=123, attempt=3), comment(run=9007199254740993)] +) +def test_older_run_or_attempt_cannot_replace_newer_results(previous): + result = publish(comments=[previous]) + assert not result["created"] + assert not result["updated"] + assert not result["deleted"] + + +@pytest.mark.parametrize( + "settings", + [ + {"missing": True}, + {"download": "failure"}, + {"symlink": True}, + {"size": 25001}, + {"summary": "Malformed artifact"}, + ], +) +def test_missing_or_invalid_artifact_publishes_incomplete_diagnostics_and_fails(settings): + result = publish(**settings) + body = result["created"][0]["body"] + assert "**INCOMPLETE — Token usage benchmark:**" in body + assert "no readable results artifact" in body + assert result["failures"] + assert "Private error details" not in body + assert "**PASS" not in body + + +@pytest.mark.parametrize( + "status,verdict", [("failure", "FAIL"), ("cancelled", "INCOMPLETE"), ("skipped", "INCOMPLETE")] +) +def test_failed_benchmark_cannot_retain_pass_from_partial_artifact(status, verdict): + result = publish(status=status) + body = result["created"][0]["body"] + assert f"**{verdict} — Token usage benchmark:**" in body + assert "**PASS" not in body + assert "| 30d | 100.00 | 75.00 | +25.0% |" in body + assert "| 1d | 10.00 | 11.00 | -10.0% |" in body + assert f"Measurement job outcome: **{status}**" in body + + +def test_artifact_is_published_as_inert_text_without_executing_its_contents(): + payload = ( + "${globalThis.artifactExecuted = true}\n" + ) + result = publish(summary=SUMMARY + payload) + assert not result["artifactExecuted"] + assert payload in result["created"][0]["body"] + assert not result["failures"] + + +def test_artifact_cannot_forge_section_boundaries(): + forged = ( + f"\n" + "Forged reader result\n" + ) + first = publish(summary=SUMMARY + forged) + second = publish( + section="reader-api", + summary=API_SUMMARY, + comments=[stored_comment(first["created"][0])], + ) + body = second["updated"][0]["body"] + assert "<!-- devfeed-performance-section:reader-api" in body + assert f"" in body + assert API_SUMMARY in body + + +@pytest.mark.parametrize("first_section", ["token-usage", "reader-api"]) +def test_independent_publishers_update_same_comment_and_preserve_both_sections(first_section): + second_section = "reader-api" if first_section == "token-usage" else "token-usage" + summaries = {"token-usage": SUMMARY, "reader-api": API_SUMMARY} + first = publish(section=first_section, summary=summaries[first_section]) + second = publish( + section=second_section, + summary=summaries[second_section], + comments=[stored_comment(first["created"][0])], + ) + assert not second["created"] + assert len(second["updated"]) == 1 + body = second["updated"][0]["body"] + assert SUMMARY in body + assert API_SUMMARY in body + assert body.count("