Add auction timeline offsets spec amendment

.gitignore

@@ -63,3 +63,4 @@ src/*.html
 # leftover local build artifacts (node_modules, target, dist) that remain on disk.
 /crates/js/
 /crates/integration-tests/
+wrangler.integration.generated.toml

crates/trusted-server-core/src/access_telemetry.rs

@@ -286,6 +286,10 @@ pub fn access_event_row(
         "stream_ms": timings.stream_ms,
         "request_elapsed_ms": timings.request_elapsed_ms,
         "resp_bytes": timings.resp_bytes,
+        "auction_dispatched_ms": timings.auction_dispatched_ms,
+        "auction_resolved_ms": timings.auction_resolved_ms,
+        "auction_committed_ms": timings.auction_committed_ms,
+        "auction_id": timings.auction_id.as_deref().unwrap_or("none"),
         "template_cache_state": snapshot.template_cache_state,
         "country": snapshot.country,
         "ts_version": snapshot.ts_version,
@@ -463,6 +467,9 @@ mod tests {
             "stream_ms",
             "request_elapsed_ms",
             "resp_bytes",
+            "auction_dispatched_ms",
+            "auction_resolved_ms",
+            "auction_committed_ms",
         ] {
             assert!(
                 parsed[field].is_null(),
@@ -488,6 +495,10 @@ mod tests {
             );
         }
         assert_eq!(parsed["auction_wait_placement"], "none");
+        assert_eq!(
+            parsed["auction_id"], "none",
+            "auction_id should carry the none sentinel when no auction ran"
+        );
     }
 
     #[test]
@@ -506,6 +517,10 @@ mod tests {
             stream_ms: Some(8),
             auction_wait_placement: Some(AuctionWaitPlacement::InStream),
             resp_bytes: Some(1024),
+            auction_dispatched_ms: Some(9),
+            auction_resolved_ms: Some(10),
+            auction_committed_ms: Some(11),
+            auction_id: Some("33333333-3333-3333-3333-333333333333".to_owned()),
         };
         let row = access_event_row(&snapshot, &timings, 1_700_000_000_000);
         let parsed: serde_json::Value =
@@ -515,6 +530,10 @@ mod tests {
         assert_eq!(parsed["stream_ms"], 8);
         assert_eq!(parsed["resp_bytes"], 1024);
         assert_eq!(parsed["auction_wait_placement"], "in_stream");
+        assert_eq!(parsed["auction_dispatched_ms"], 9);
+        assert_eq!(parsed["auction_resolved_ms"], 10);
+        assert_eq!(parsed["auction_committed_ms"], 11);
+        assert_eq!(parsed["auction_id"], "33333333-3333-3333-3333-333333333333");
     }
 
     #[test]

crates/trusted-server-core/src/publisher.rs

@@ -3951,6 +3951,8 @@ async fn collect_non_html_auction(
     params
         .timings
         .record_auction_wait(placement, wait_started.elapsed());
+    // T0-anchored timeline mark (spec section 18): final bid or timeout.
+    params.timings.mark_auction_resolved();
     let delivered_winner_slots = write_bids_to_state(
         &result.winning_bids,
         params.price_granularity,
@@ -3960,6 +3962,9 @@ async fn collect_non_html_auction(
         settings.debug.inject_adm_for_testing,
         auction_id.as_deref(),
     );
+    // T0-anchored timeline mark (spec section 18): winning bids are in page
+    // state, available to the response pipeline.
+    params.timings.mark_auction_committed();
     if let (Some(observation), Some(auction_request)) =
         (telemetry.observation, telemetry.auction_request.as_ref())
     {
@@ -4010,6 +4015,8 @@ async fn collect_stream_auction(
         .collect_dispatched_auction(dispatched, services, &collect_ctx)
         .await;
     timings.record_auction_wait(*placement, wait_started.elapsed());
+    // T0-anchored timeline mark (spec section 18): final bid or timeout.
+    timings.mark_auction_resolved();
     log::info!(
         "body_close_hold_loop: collect complete - {} winning bid(s)",
         result.winning_bids.len()
@@ -4023,6 +4030,9 @@ async fn collect_stream_auction(
         settings.debug.inject_adm_for_testing,
         auction_id.as_deref(),
     );
+    // T0-anchored timeline mark (spec section 18): winning bids are in page
+    // state, available to the response pipeline.
+    timings.mark_auction_committed();
     if let (Some(observation), Some(auction_request)) =
         (telemetry.observation, telemetry.auction_request.as_ref())
     {
@@ -4340,6 +4350,10 @@ pub async fn handle_publisher_request(
                 .await
             {
                 DispatchAuctionOutcome::Dispatched(dispatched) => {
+                    // T0-anchored timeline mark (spec section 18): bid
+                    // requests have left the edge. A failed dispatch never
+                    // marks, so all three auction offsets stay null for it.
+                    timings.mark_auction_dispatched(observation.auction_id.to_string());
                     auction_request_for_telemetry = Some(auction_request);
                     auction_observation = Some(observation);
                     Some(dispatched)

crates/trusted-server-core/src/request_timing.rs

@@ -106,6 +106,19 @@ struct Inner {
     /// Response body size in bytes, set via
     /// [`RequestTimings::set_resp_bytes`].
     resp_bytes: Option<u64>,
+    /// Elapsed time at the first
+    /// [`RequestTimings::mark_auction_dispatched`] call.
+    auction_dispatched: Option<Duration>,
+    /// Elapsed time at the first
+    /// [`RequestTimings::mark_auction_resolved`] call.
+    auction_resolved: Option<Duration>,
+    /// Elapsed time at the first
+    /// [`RequestTimings::mark_auction_committed`] call.
+    auction_committed: Option<Duration>,
+    /// Telemetry auction UUID recorded by the first
+    /// [`RequestTimings::mark_auction_dispatched`] call; joins the access
+    /// row to the per-bidder auction dataset.
+    auction_id: Option<String>,
 }
 
 /// Per-request phase timing collector.
@@ -128,6 +141,10 @@ impl RequestTimings {
             request_elapsed: None,
             auction_wait_placement: None,
             resp_bytes: None,
+            auction_dispatched: None,
+            auction_resolved: None,
+            auction_committed: None,
+            auction_id: None,
         })))
     }
 
@@ -200,6 +217,53 @@ impl RequestTimings {
         }
     }
 
+    /// Stamps the elapsed time since `t0` as the auction dispatch offset and
+    /// records the telemetry auction id, the first time this is called.
+    ///
+    /// Called when `dispatch_auction` reports the bid requests dispatched;
+    /// a failed dispatch never records, so all three auction offsets stay
+    /// `None` for it. Subsequent calls are no-ops (first call wins). Drops
+    /// the sample silently on lock contention or poisoning.
+    pub fn mark_auction_dispatched(&self, auction_id: String) {
+        let Ok(mut inner) = self.0.try_lock() else {
+            return;
+        };
+        if inner.auction_dispatched.is_none() {
+            inner.auction_dispatched = Some(inner.t0.elapsed());
+            inner.auction_id = Some(auction_id);
+        }
+    }
+
+    /// Stamps the elapsed time since `t0` as the auction resolve offset (the
+    /// final bid returned or the auction timed out), the first time this is
+    /// called.
+    ///
+    /// Subsequent calls are no-ops (first call wins). Drops the sample
+    /// silently on lock contention or poisoning.
+    pub fn mark_auction_resolved(&self) {
+        let Ok(mut inner) = self.0.try_lock() else {
+            return;
+        };
+        if inner.auction_resolved.is_none() {
+            inner.auction_resolved = Some(inner.t0.elapsed());
+        }
+    }
+
+    /// Stamps the elapsed time since `t0` as the auction commit offset
+    /// (winning bids written into page state), the first time this is
+    /// called.
+    ///
+    /// Subsequent calls are no-ops (first call wins). Drops the sample
+    /// silently on lock contention or poisoning.
+    pub fn mark_auction_committed(&self) {
+        let Ok(mut inner) = self.0.try_lock() else {
+            return;
+        };
+        if inner.auction_committed.is_none() {
+            inner.auction_committed = Some(inner.t0.elapsed());
+        }
+    }
+
     /// Records the response body size in bytes.
     ///
     /// Drops the sample silently on lock contention or poisoning.
@@ -257,6 +321,10 @@ impl RequestTimings {
             stream_ms: duration_ms(inner.phases[Phase::Stream.index()]),
             auction_wait_placement: inner.auction_wait_placement,
             resp_bytes: inner.resp_bytes,
+            auction_dispatched_ms: duration_ms(inner.auction_dispatched),
+            auction_resolved_ms: duration_ms(inner.auction_resolved),
+            auction_committed_ms: duration_ms(inner.auction_committed),
+            auction_id: inner.auction_id.clone(),
         }
     }
 }
@@ -379,6 +447,18 @@ pub struct TimingSnapshot {
     /// Response body size in bytes, set via
     /// [`RequestTimings::set_resp_bytes`].
     pub resp_bytes: Option<u64>,
+    /// T0 offset at which the auction dispatched (bid requests left the
+    /// edge), or `None` when no auction ran.
+    pub auction_dispatched_ms: Option<u32>,
+    /// T0 offset at which the auction resolved (final bid or timeout), or
+    /// `None` when no auction ran.
+    pub auction_resolved_ms: Option<u32>,
+    /// T0 offset at which winning bids were committed into page state, or
+    /// `None` when no auction ran.
+    pub auction_committed_ms: Option<u32>,
+    /// Telemetry auction UUID joining this row to the auction dataset, or
+    /// `None` when no auction ran.
+    pub auction_id: Option<String>,
 }
 
 #[cfg(test)]
@@ -419,6 +499,60 @@ mod tests {
         );
     }
 
+    #[test]
+    fn auction_marks_are_first_call_wins_and_snapshot_maps_them() {
+        let timings = RequestTimings::new();
+        timings.mark_auction_dispatched("11111111-1111-1111-1111-111111111111".to_owned());
+        timings.mark_auction_resolved();
+        timings.mark_auction_committed();
+        // Second calls must not overwrite the first-recorded values.
+        timings.mark_auction_dispatched("22222222-2222-2222-2222-222222222222".to_owned());
+        timings.mark_auction_resolved();
+        timings.mark_auction_committed();
+
+        let snapshot = timings.snapshot();
+        assert!(
+            snapshot.auction_dispatched_ms.is_some(),
+            "should record the dispatch offset"
+        );
+        assert!(
+            snapshot.auction_resolved_ms.is_some(),
+            "should record the resolve offset"
+        );
+        assert!(
+            snapshot.auction_committed_ms.is_some(),
+            "should record the commit offset"
+        );
+        assert_eq!(
+            snapshot.auction_id.as_deref(),
+            Some("11111111-1111-1111-1111-111111111111"),
+            "should keep the first-recorded auction id"
+        );
+    }
+
+    #[test]
+    fn snapshot_without_auction_marks_yields_none_for_all_offsets() {
+        let timings = RequestTimings::new();
+        timings.mark_headers_ready();
+        let snapshot = timings.snapshot();
+        assert_eq!(
+            snapshot.auction_dispatched_ms, None,
+            "should stay None when no auction dispatched"
+        );
+        assert_eq!(
+            snapshot.auction_resolved_ms, None,
+            "should stay None when no auction resolved"
+        );
+        assert_eq!(
+            snapshot.auction_committed_ms, None,
+            "should stay None when no auction committed"
+        );
+        assert_eq!(
+            snapshot.auction_id, None,
+            "should carry no auction id when no auction ran"
+        );
+    }
+
     #[test]
     fn render_omits_unrecorded_phases_and_orders_total_first() {
         let timings = RequestTimings::new();

docs/superpowers/plans/2026-08-26-auction-timeline-offsets.md

@@ -0,0 +1,59 @@
+# Auction Timeline Offsets Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Record three T0-anchored auction milestones (dispatched, resolved, committed) plus the auction id on `RequestTimings`, and emit them as four additive columns on the `access_logs_raw` row.
+
+**Architecture:** Follows spec section 18 exactly. All state lives in the existing `RequestTimings` inner (same `try_lock`/first-call-wins/saturating model as `mark_headers_ready`); the row builder reads the values from `TimingSnapshot`, so no new emission path and no adapter changes.
+
+**Tech Stack:** Rust (core crate only), Tinybird datasource file.
+
+**Spec:** `docs/superpowers/specs/2026-08-24-request-phase-timing-design.md` section 18.
+
+## Global Constraints
+
+- Marks are first-call-wins; `try_lock` only; a contended lock drops the sample.
+- Null offsets mean "no auction ran", never zero. `auction_id` sentinel is `none`.
+- Column names: `auction_dispatched_ms`, `auction_resolved_ms`, `auction_committed_ms`, `auction_id`; JSONPaths `json:$.<name>`; FORWARD_QUERY extended in the same order.
+- Dispatch mark records only on `DispatchAuctionOutcome::Dispatched`; a failed dispatch leaves all three offsets null (the auction dataset still records the failure).
+- No header emission, no config surface, no changes outside `trusted-server-core` and `tinybird/`.
+
+---
+
+### Task 1: RequestTimings marks and snapshot fields
+
+**Files:**
+- Modify: `crates/trusted-server-core/src/request_timing.rs`
+
+**Interfaces:**
+- Produces: `mark_auction_dispatched(&self, auction_id: String)`, `mark_auction_resolved(&self)`, `mark_auction_committed(&self)`; `TimingSnapshot { auction_dispatched_ms, auction_resolved_ms, auction_committed_ms: Option<u32>, auction_id: Option<String>, .. }`
+
+- [ ] Add `auction_dispatched`, `auction_resolved`, `auction_committed: Option<Duration>` and `auction_id: Option<String>` to `Inner`; initialize `None`.
+- [ ] Add the three mark methods, first-call-wins on their own field, storing `inner.t0.elapsed()`; dispatched also stores the id first-call-wins.
+- [ ] Map all four into `TimingSnapshot` via `duration_ms` / clone.
+- [ ] Tests: first-call-wins per mark; snapshot maps offsets and id; unmarked snapshot yields all `None`.
+- [ ] `cargo test-fastly request_timing`, commit.
+
+### Task 2: Publisher call sites
+
+**Files:**
+- Modify: `crates/trusted-server-core/src/publisher.rs`
+
+**Interfaces:**
+- Consumes: Task 1 methods; `observation.auction_id` (`AuctionObservationContext`), in scope at the dispatch site.
+
+- [ ] In the `DispatchAuctionOutcome::Dispatched` arm (~line 4341): `timings.mark_auction_dispatched(observation.auction_id.to_string());`
+- [ ] After both `record_auction_wait` calls (collect sites ~3952 and ~4012): `mark_auction_resolved()`.
+- [ ] After both `write_bids_to_state` calls (~3954 and ~4017): `mark_auction_committed()`.
+- [ ] `cargo test-fastly`, commit.
+
+### Task 3: Row columns and datasource
+
+**Files:**
+- Modify: `crates/trusted-server-core/src/access_telemetry.rs`
+- Modify: `tinybird/datasources/access_logs_raw.datasource`
+
+- [ ] `access_event_row`: add the three offset keys (nullable) and `auction_id` with `none` sentinel, after the existing phase keys.
+- [ ] Extend `row_serializes_nulls_for_missing_phases` and `row_serializes_recorded_phases_as_numbers` for the new keys.
+- [ ] Datasource: four schema columns with JSONPaths (`Nullable(UInt32)` ×3, `String`), appended at the end of SCHEMA and FORWARD_QUERY so existing column order stays stable.
+- [ ] Full gates: fmt, clippy (all six), test-fastly/axum/cloudflare/spin, parity. Commit.