Index topic workflow calls for bounded reporting

benchmarks/topic-report-history.md

@@ -0,0 +1,138 @@
+# Topic report history profiling
+
+Issue [#299](https://github.com/abhi1693/devfeed.tech/issues/299) concerns repeatedly
+expanding large decision state JSON and sorting repeated stages in administrative
+reports. The disposable profiler compares the complete previous and current
+decision metrics and inference chart responses on the same fixture. It alternates
+strategy order and requires exact JSON equality for 1-, 7-, 30- and 90-day reports.
+
+`topic_decision_calls` stores one typed fact for each original `(proposal_id,
+ordinal)`, including its lifetime occurrence number within its stage. Source run
+JSON remains authoritative. A transactional trigger filters unchanged facts before
+the upsert, avoiding conflict locks on unaffected calls, and deletes removed tail
+entries; unrelated state updates skip projection work. Each
+write handles that run's calls, with no global rollup or separate refresh scheduler.
+Report time filters use the fact timestamp index, while completed proposal usage
+joins all lifetime facts for proposals reviewed in the requested window.
+
+Migration `0036` follows the scheduler clock migration `0035`, installs the topic
+projection, backfills existing runs and creates the indexes atomically. Its bounded
+lock admission waits at most 30 seconds before a retryable failure. Once admitted,
+backfill/index creation pauses proposal and run writes while allowing ordinary reads.
+The current application requires schema `0036`; rollback application code together
+with the schema. Downgrading to `0035` removes the derived topic facts while preserving
+source JSON and the scheduler clock changes, allowing a later upgrade to rebuild the facts.
+Downgrade briefly takes exclusive table locks, waiting for active readers and
+writers before removing the trigger and projection; schedule it during a quiet
+period.
+
+The production reference was collected read-only on 24 October 2026 at 09:29 UTC:
+92,581 articles, 6,011 proposals, 5,000 decision runs and 14,373 calls. PostgreSQL
+17.11 ran natively on ARM64 with `work_mem=4MB`. A bounded 100-run sample had mean
+state JSON size 20,511 bytes, median 22,131 and p95 35,444 bytes. These cardinalities
+describe that observation, rather than a continuously refreshed production count.
+
+`scripts/profile_topic_reports.py` uses migrated, empty, explicitly local
+PostgreSQL 17+ databases ending in `_test`. It refuses libpq routing overrides and
+never reads the checkout's `.env` or truncates fixture tables. The benchmark uses
+ordinary article, inference and decision triggers during seeding. Each run has
+approximately 20 KiB of incompressible unrelated evidence, plus one to five ordered
+calls with repeated draft and verification stages spaced six hours apart and offset
+by one hour from exact integer-day boundaries. Repeated stages span UTC dates and
+the 30- and 90-day report boundaries. The first 6,000 runs cover the
+recent 90 days; further runs add older unrelated history. Proposal review dates
+vary independently, exercising whole-workflow accounting beyond the report window.
+
+The ARM64 workflow runs 100,000 and 300,000 article fixtures, with respectively
+6,000 and 18,000 decision runs and 500,000 and 1,500,000 inference ledger rows.
+It saves complete `EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON)` plans, report wall
+time, application process CPU, shared buffers and temporary blocks as artifacts.
+Four concurrent workers run 90-day report pairs alongside 200 article insert and decision
+correction transactions, rotating through one- to five-call workflows. Both
+strategies execute the same operations, paced identically using the measured legacy
+runtime to keep the writer phase active throughout reporting. The artifact records
+`writer_phase_overlap` for every report. That phase includes SQL, rollback and pacing
+sleeps; the event does not prove that every report overlaps an actual writer SQL
+statement. Transaction latency excludes pacing. These writes
+roll back so report inputs remain identical; ingestion latency measures transaction
+work without durable commit/fsync cost. Concurrent process CPU is reported once for
+the entire process; summing per-report CPU under concurrency would double count.
+
+The legacy concurrency phase briefly disables only the new decision projection
+trigger on this dedicated fixture database, restoring it in `finally`. Committing
+that short trigger-state change prevents an open `ALTER TABLE` transaction from
+holding an exclusive lock during reporting. No workload writes commit. A separate
+rolled-back five-call correction captures before/after WAL and trigger execution after a
+warm-up sample avoids charging only the first strategy for checkpoint-dependent
+full-page images. The root `EXPLAIN` WAL excludes AFTER-trigger work, so
+`statement_wal_bytes` also records the server-wide LSN delta around the statement,
+including trigger WAL. This dedicated-server measurement can include background
+WAL and excludes the later transaction rollback. A full
+equivalent backfill measures initial extraction and index insertion separately from
+steady-state reporting. An equivalent backfill excludes migration DDL/lock wait.
+
+Separate committed-write comparisons run after the read diagnostics. Fresh cohorts
+start with four or nineteen calls, then append a fifth or twentieth call, settle it,
+correct an earlier call and save an unrelated checkpoint. These are four separate
+transactions per workflow; the append also inserts an article. Latency ends after
+COMMIT. Five calls cover the configured default; twenty cover the maximum initial
+budget plus both operator grants. Each strategy uses the same logical source states
+and 20 KiB of unrelated evidence. Repeated pairs alternate strategy order. Final
+source-state hashes must match, and all current facts must match their source before
+cleanup. Setup, trigger changes, verification and cleanup are outside timing.
+
+Committed samples report per-operation median and overall p95, transaction counts,
+application thread CPU and WAL snapshots. They measure these lifecycle transactions,
+not the full worker pipeline or maximum ingestion capacity. Each phase uses fresh
+rows on the same dedicated database; cleanup does not restore identical heap/index
+bloat. Server-wide WAL counters include background work and can contain unequal
+full-page images. Retain those counts rather than interpreting one ratio as stable
+write amplification. No universal acceptable writer-latency budget has been defined.
+
+To run locally, create and migrate a new disposable database and make a clean
+checkout of the pre-change revision. Set `DEVFEED_DATABASE_URL` to that database
+for migrations, and export the same URL as `DEVFEED_TEST_DATABASE_URL` for profiling:
+
+```sh
+uv run --locked --no-sync alembic upgrade head
+uv run --locked --no-sync python -m scripts.profile_topic_reports \
+  --baseline-root /path/to/baseline --seed --articles 100000 \
+  --output reports/topic-reports.json
+```
+
+Output must be a new `.json` artifact under the invocation's `reports/` directory.
+Absolute paths inside that directory also work. Traversal components, symlink files
+or directories, and existing destinations are rejected before database work. Choose
+a new filename for each local run; artifact creation never overwrites an earlier report.
+
+The path-filtered `Topic report history profile` pull-request workflow uses a native
+`ubuntu-24.04-arm` runner and an immutable PostgreSQL 17.11 image, explicitly requiring
+major version 17 with `--require-postgres-major 17`. Profiling has read-only repository
+permissions and no deployment or production credentials. Local AMD64
+smoke measurements validate the harness; they do not establish native ARM64 speed.
+When the pull-request base still uses historical JSON expansion, the workflow
+requires lower concurrent report and topic SQL runtime, zero topic SQL temporary
+writes, and complete overlap between reports and the paced writer phase. Later changes compare
+against their current base and retain exact output parity without requiring every
+change to make the already normalized queries faster again. Full phase
+elapsed time includes identical writer pacing and is not a report latency metric.
+These are read checks, not an overall performance verdict. The summary reports
+committed write costs separately and marks them unverified when absent; a passing
+read check cannot establish acceptable long-term write overhead.
+
+Each run publishes separate 100,000- and 300,000-article before/after tables with
+signed cost deltas and improvement percentages in its job summary and summary artifact.
+Positive improvement means reduced cost; negative improvement means increased cost.
+Same-repository pull requests
+also receive one updated bot comment. The publishing job executes no checkout or
+artifact code, validates the current attempt's JSON as data, and checks the current
+PR head before writing. Forks retain the summary and artifact. Failed jobs and
+missing or incomplete artifacts are reported explicitly instead of reusing a prior
+attempt's passing measurement.
+
+After deployment, compare fresh `pg_stat_statements` deltas under matched report
+windows and concurrency: calls, total execution time, temporary blocks per call and
+buffer reads. Also check report freshness, worker latency and write amplification,
+then verify a retried correction and deletion converge immediately without duplicate
+facts. These deployment checks require a separately authorized release; the
+benchmark workflow only accesses its disposable database.

migrations/versions/0036_topic_decision_calls.py

@@ -0,0 +1,134 @@
+"""Project topic call facts once, preserving lifetime stage and token accounting."""
+
+from alembic import op
+
+revision = "0036"
+down_revision = "0035"
+branch_labels = None
+depends_on = None
+
+
+def _admit_locks(*, downgrading=False):
+    # These fixed modes cover every relation lock needed by the subsequent DDL.
+    # A failed attempt releases all partial locks before waiting, including when
+    # a writer flushes run -> proposal or a report reads facts -> run/proposal.
+    locks = (
+        """LOCK TABLE topic_proposals IN ACCESS EXCLUSIVE MODE NOWAIT;
+              LOCK TABLE topic_decision_runs IN ACCESS EXCLUSIVE MODE NOWAIT;
+              LOCK TABLE topic_decision_calls IN ACCESS EXCLUSIVE MODE NOWAIT;"""
+        if downgrading
+        else """LOCK TABLE topic_proposals IN SHARE MODE NOWAIT;
+              LOCK TABLE topic_decision_runs IN EXCLUSIVE MODE NOWAIT;"""
+    )
+    op.execute(f"""DO $body$
+        DECLARE deadline timestamptz := clock_timestamp() + interval '30 seconds';
+        BEGIN
+          LOOP
+            BEGIN
+              {locks}
+              EXIT;
+            EXCEPTION WHEN lock_not_available THEN
+              IF clock_timestamp() >= deadline THEN RAISE; END IF;
+            END;
+            PERFORM pg_sleep(0.05);
+          END LOOP;
+        END $body$""")
+
+
+def upgrade():
+    # Exclude run row lockers: backfill's foreign-key checks acquire row locks.
+    # Reads continue; source writes pause once both locks are held through commit.
+    _admit_locks()
+    op.execute("""CREATE TABLE topic_decision_calls (
+        proposal_id uuid NOT NULL REFERENCES topic_decision_runs(proposal_id) ON DELETE CASCADE,
+        ordinal bigint NOT NULL, stage text, stage_call bigint NOT NULL,
+        started_at timestamptz, tokens numeric NOT NULL,
+        charged_tokens bigint, escalated boolean NOT NULL,
+        PRIMARY KEY (proposal_id, ordinal))""")
+    # One projection is shared by backfill and live writes. Scalar array entries
+    # still occupy their original ordinal and retain an unknown token contribution.
+    # Numeric addition avoids overflowing when individually valid bigint fields add up.
+    op.execute("""CREATE FUNCTION topic_decision_call_values(call_array jsonb)
+        RETURNS TABLE (ordinal bigint, stage text, stage_call bigint, started_at timestamptz,
+            tokens numeric, charged_tokens bigint, escalated boolean)
+        LANGUAGE sql STABLE SET TimeZone = 'UTC' AS $body$
+        WITH calls AS (
+          SELECT c, n AS ordinal FROM jsonb_array_elements(
+            CASE WHEN jsonb_typeof(call_array) = 'array' THEN call_array ELSE '[]'::jsonb END
+          ) WITH ORDINALITY AS items(c, n)
+        ), parsed AS (
+          SELECT *,
+            CASE WHEN pg_input_is_valid(c#>>'{tokens,totalTokens}', 'bigint')
+              THEN (c#>>'{tokens,totalTokens}')::numeric ELSE 0 END AS total_tokens,
+            CASE WHEN pg_input_is_valid(c#>>'{tokens,inputTokens}', 'bigint')
+              THEN (c#>>'{tokens,inputTokens}')::numeric ELSE 0 END AS input_tokens,
+            CASE WHEN pg_input_is_valid(c#>>'{tokens,outputTokens}', 'bigint')
+              THEN (c#>>'{tokens,outputTokens}')::numeric ELSE 0 END AS output_tokens
+          FROM calls
+        )
+        SELECT ordinal, c->>'stage',
+          row_number() OVER (PARTITION BY c->>'stage' ORDER BY ordinal),
+          -- Relative and infinite timestamps would move historical facts on a later
+          -- correction or rebuild. Only finite ISO dates/date-times are telemetry.
+          CASE WHEN c->>'started_at' ~ (
+            '^[0-9]{4}-[0-9]{2}-[0-9]{2}([Tt ][0-9]{2}:[0-9]{2}' ||
+            '(:[0-9]{2}([.][0-9]+)?)?([Zz]|[+-][0-9]{2}(:?[0-9]{2})?)?)?$')
+            AND pg_input_is_valid(c->>'started_at', 'timestamptz')
+            THEN (c->>'started_at')::timestamptz END,
+          greatest(total_tokens, input_tokens + output_tokens),
+          CASE WHEN pg_input_is_valid(c->>'charged_tokens', 'bigint')
+            THEN (c->>'charged_tokens')::bigint END,
+          CASE WHEN pg_input_is_valid(c->>'escalated', 'boolean')
+            THEN (c->>'escalated')::boolean ELSE false END
+        FROM parsed
+        $body$""")
+    op.execute("""INSERT INTO topic_decision_calls
+        SELECT r.proposal_id, calls.* FROM topic_decision_runs r
+        CROSS JOIN LATERAL topic_decision_call_values(r.state->'calls') calls""")
+    op.execute("""CREATE FUNCTION topic_decision_calls_sync() RETURNS trigger
+        LANGUAGE plpgsql AS $body$
+        DECLARE call_count integer;
+        BEGIN
+          IF TG_OP = 'UPDATE' AND OLD.state->'calls' IS NOT DISTINCT FROM NEW.state->'calls' THEN
+            RETURN NULL;
+          END IF;
+          INSERT INTO topic_decision_calls AS target
+          SELECT NEW.proposal_id, calls.* FROM topic_decision_call_values(NEW.state->'calls') calls
+          LEFT JOIN topic_decision_calls existing
+            ON existing.proposal_id = NEW.proposal_id AND existing.ordinal = calls.ordinal
+          -- ON CONFLICT locks even unchanged rows before checking its WHERE clause.
+          -- The source run row serializes writers; skip those conflicts entirely.
+          WHERE existing.ordinal IS NULL OR
+            (existing.stage, existing.stage_call, existing.started_at,
+             existing.tokens, existing.charged_tokens, existing.escalated)
+            IS DISTINCT FROM (calls.stage, calls.stage_call, calls.started_at,
+                              calls.tokens, calls.charged_tokens, calls.escalated)
+          ON CONFLICT (proposal_id, ordinal) DO UPDATE SET
+            (stage, stage_call, started_at, tokens, charged_tokens, escalated) =
+            (excluded.stage, excluded.stage_call, excluded.started_at,
+             excluded.tokens, excluded.charged_tokens, excluded.escalated)
+          WHERE (target.stage, target.stage_call, target.started_at,
+                 target.tokens, target.charged_tokens, target.escalated)
+            IS DISTINCT FROM (excluded.stage, excluded.stage_call, excluded.started_at,
+                              excluded.tokens, excluded.charged_tokens, excluded.escalated);
+          call_count := CASE WHEN jsonb_typeof(NEW.state->'calls') = 'array'
+            THEN jsonb_array_length(NEW.state->'calls') ELSE 0 END;
+          DELETE FROM topic_decision_calls
+            WHERE proposal_id = NEW.proposal_id AND ordinal > call_count;
+          RETURN NULL;
+        END $body$""")
+    op.execute("""CREATE TRIGGER topic_decision_calls_sync
+        AFTER INSERT OR UPDATE OF state ON topic_decision_runs
+        FOR EACH ROW EXECUTE FUNCTION topic_decision_calls_sync()""")
+    # The FK propagates deletion and TRUNCATE ... CASCADE without a second writer.
+    op.create_index("ix_topic_decision_calls_started_at", "topic_decision_calls", ["started_at"])
+    op.create_index("ix_topic_proposals_reviewed_at", "topic_proposals", ["reviewed_at"])
+
+
+def downgrade():
+    _admit_locks(downgrading=True)
+    op.drop_index("ix_topic_proposals_reviewed_at", table_name="topic_proposals")
+    op.execute("DROP TRIGGER topic_decision_calls_sync ON topic_decision_runs")
+    op.execute("DROP FUNCTION topic_decision_calls_sync()")
+    op.execute("DROP FUNCTION topic_decision_call_values(jsonb)")
+    op.drop_table("topic_decision_calls")

packages/core/src/inference_metrics.py

@@ -161,8 +161,9 @@ def inference_charts(session, start: datetime, now: datetime) -> InferenceCharts
         else _ledger_activity(session, start, now)
     )
     days = {}
-    day = start.date()
-    while day <= now.date():
+    day = (start.replace(tzinfo=UTC) if start.tzinfo is None else start.astimezone(UTC)).date()
+    last_day = (now.replace(tzinfo=UTC) if now.tzinfo is None else now.astimezone(UTC)).date()
+    while day <= last_day:
         key = day.isoformat()
         days[key] = TopicDecisionDay(date=key)
         day += timedelta(days=1)
@@ -179,17 +180,11 @@ def inference_charts(session, start: datetime, now: datetime) -> InferenceCharts
         days[row["date"]].approved, days[row["date"]].rejected = row["approved"], row["rejected"]
     calls = session.execute(
         text("""
-        WITH calls AS (
-          SELECT proposal_id, c, n,
-            row_number() OVER (PARTITION BY proposal_id, c->>'stage' ORDER BY n) AS stage_call
-          FROM topic_decision_runs,
-            jsonb_array_elements(state->'calls') WITH ORDINALITY AS items(c, n)
-        )
-        SELECT to_char((c->>'started_at')::timestamptz AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS date,
+        SELECT to_char(started_at AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS date,
           count(*) AS calls,
-          count(*) FILTER (WHERE (c->>'escalated')::boolean) AS escalations,
+          count(*) FILTER (WHERE escalated) AS escalations,
           count(*) FILTER (WHERE stage_call > 1) AS repeated_stages
-        FROM calls WHERE (c->>'started_at')::timestamptz BETWEEN :start AND :now GROUP BY date
+        FROM topic_decision_calls WHERE started_at BETWEEN :start AND :now GROUP BY date
     """),
         {"start": start, "now": now},
     ).mappings()