> ## Documentation Index
> Fetch the complete documentation index at: https://docs.perplexity.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Track What the Web Said Each Week About a Topic, and Build a Report

> Search engines tell you what the web says today. This recipe shows what was being reported about a topic during each week of a date range: the Agent API searches each week with publication-date filters, explains how the story changed, and counts the outlets that covered it, while your code writes the timeline and draws the chart.

Suppose you need to know how coverage of a topic changed week by week. An ordinary search does not help, because results are ranked for today. In a probe for this recipe, a question about one September week, sent without date filters, returned 19 results: only 3 were from that week, and 9 were from the day the probe ran. Anyone following a product launch, a competitor, or a regulation ends up rebuilding the week-by-week picture by hand.

With the Agent API, this becomes a short script:

* **Date-filtered search.** `web_search` accepts publication-date filters. One request per week returns only pages published that week, with citations.
* **Effort levels instead of model and prompt choices.** [Effort levels](https://docs.perplexity.ai/docs/agent-api/effort.md) set how much work the model does. `effort="fast"` runs one quick search step, right for a narrow dated question. `effort="medium"` gives the model more steps to read all the weeks and explain how the story changed.
* **Code the model writes and runs.** A third request adds the `sandbox` tool. The model writes Python to count which outlets covered each week, and the API runs it.
* **Your code does the rest.** It splits the range into weeks, runs the fast requests in parallel, and writes the timeline and the chart.

You get a Markdown timeline with a cited summary for every week, a narrative of how the coverage changed with the turning points named, a table of outlets per week, and a chart.

```bash theme={null}
python timeline.py "iPhone 18 Pro" --from 2026-09-01 --to 2026-10-26 --step 7
```

Here is what came back for the iPhone 18 Pro, September 1 to October 26. First the `Week by week` block the script prints, then the turning points the synthesis named.

```text theme={null}
Sep 01 to Sep 07  iPhone 18 Pro expected to debut September 9
Sep 08 to Sep 14  Apple unveiled iPhone 18 Pro with variable-aperture camera
Sep 15 to Sep 21  iPhone 18 Pro launched with upgraded camera and chip
Sep 22 to Sep 28  Face ID glitches reported on some iPhone 18 Pro models
Sep 29 to Oct 05  Variable-aperture camera reaches iPhone 18 Pro video
Oct 06 to Oct 12  Apple reportedly cut iPhone 18 Pro component orders
Oct 13 to Oct 19  Reports diverged on iPhone 18 Pro demand
Oct 20 to Oct 26  Apple expanded U.S. manufacturing partnerships for iPhone 18 Pro

Turning points
Sep 08 to Sep 14  Rumor becomes product
Sep 22 to Sep 28  Early reliability scrutiny
Oct 06 to Oct 12  Demand narrative fractures
Oct 20 to Oct 26  Supply and repair take focus
```

Each headline is the model's one-line answer for that week, written from pages dated inside it.

<video autoPlay muted loop playsInline controls src="https://mintcdn.com/perplexity/tQKmFpZHR3huTLSd/docs/assets/images/cookbook/examples/agent-api-story-timeline-demo.mp4?fit=max&auto=format&n=tQKmFpZHR3huTLSd&q=85&s=8023204c53f16cead7c7ce4382a77a35" />

One recorded run at 1.5x speed, no cuts. The terminal prints each week as its fast request finishes, then streams the synthesis while the sandbox step runs. The left pane fills from the run's own output, then shows the turning points, the sandbox table, and the chart your code draws.

![Bar chart titled iPhone 18 Pro: sources per window and turning points. Eight bars for window starts from Sep 01 to Oct 20 with heights 13, 14, 13, 15, 17, 15, 19 and 20 distinct dated sources. Numbered red markers sit on the Sep 08, Sep 22, Oct 06 and Oct 20 bars, with a key below reading 1. Sep 08: Rumor becomes product, 2. Sep 22: Early reliability scrutiny, 3. Oct 06: Demand narrative fractures, 4. Oct 20: Supply and repair take focus.](https://mintcdn.com/perplexity/tQKmFpZHR3huTLSd/docs/assets/images/cookbook/examples/agent-api-story-timeline.png?fit=max\&auto=format\&n=tQKmFpZHR3huTLSd\&q=85\&s=65c83234b9e888a4284657476acf348c)

The chart the script wrote. Bars: distinct dated sources per week, capped at the 20 results requested per call. Markers: the weeks the synthesis named as turning points. The model never sees this chart.

The recorded run cost about five cents: $0.0141 for the eight fast requests, $0.0015 for the synthesis, and $0.0323 for the sandbox step ($0.03 of it is the sandbox session).

The rest of this page says "window" rather than "week", because `--step` lets a window be a day or a month.

## Filter by date instead of asking in the prompt

Putting the dates in the prompt does not work. The search tool never sees your prompt. It ranks pages by relevance and recency, so a prompt about "the week of September 15" still gets today's pages back. In the unfiltered probe above, the model had only the 3 pages that happened to fall in that week. With the date filters on, the same window returned 17 pages dated inside it.

The date filters decide which pages the model sees. The prompt decides what it does with them.

`web_search` has four date filters, all in `MM/DD/YYYY` format: `search_after_date_filter` and `search_before_date_filter` on the publication date, `last_updated_after_filter` and `last_updated_before_filter` on the last update. This recipe uses the publication pair.

> [!NOTE]
>
> Both bounds were inclusive in every probe and the recorded run: results dated on the filter day itself came back. The [web search filter reference](https://docs.perplexity.ai/docs/agent-api/tools/web-search.md#recency-filter) is the source of truth if that changes.

## Check what you need

* Python 3.12 on macOS or Linux. The shell commands use Unix syntax.
* A Perplexity API key in `PERPLEXITY_API_KEY`. The `perplexity` library reads it on its own.
* About five cents of API credit per eight-window run, three of them for the sandbox session.

The recipe never names a model. It picks an [effort level](https://docs.perplexity.ai/docs/agent-api/effort.md) for each request instead, and the API chooses the model and tools behind it. `fast` runs one model step with one search, enough for a narrow dated question. `medium` runs several steps with its own searches, which the synthesis needs; the coverage step uses `medium` too and adds the `sandbox` tool. Each response reports the exact version it ran, such as `fast@2026-10-16.0`. Pass that string as `effort` to get the same behavior next time.

## Set up the project

Create a directory, a virtual environment, and the dependency file. matplotlib draws the chart and pytest runs the offline tests.

```bash theme={null}
mkdir story-timeline && cd story-timeline
python3.12 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
```

```text requirements.txt theme={null}
perplexityai==0.46.1
matplotlib==3.11.2
pytest==9.1.1
```

You will build four files:

* `timeline.py` makes the window and synthesis requests.
* `coverage.py` makes the sandbox request.
* `report.py` turns everything into Markdown, JSON, and a PNG.
* `compare.py` re-runs one window at several effort levels.

Each section below explains one piece of a file, then shows it. The full files are under "Save the full files".

## Split the range into windows

`split_windows` walks from `--from` to `--to` in steps of `--step` days. Both ends of each window are inclusive, and the last window is as short as it needs to be. Each `Window` formats itself as a filter dictionary, because the API wants `MM/DD/YYYY`, not ISO dates.

> **Read Window, mdy, and split\_windows (timeline.py)**
>
> ```python theme={null}
> @dataclass
> class Window:
>     index: int
>     start: date
>     end: date
>
>     @property
>     def label(self) -> str:
>         return f"{self.start:%b %d} to {self.end:%b %d}"
>
>     def filters(self) -> dict:
>         return {
>             "search_after_date_filter": mdy(self.start),
>             "search_before_date_filter": mdy(self.end),
>         }
>
>
> def mdy(d: date) -> str:  # date filters take MM/DD/YYYY, not ISO
>     return f"{d.month:02d}/{d.day:02d}/{d.year}"
>
>
> def split_windows(start: date, end: date, step: int) -> list[Window]:
>     """Inclusive windows of `step` days; the last one is as short as it needs to be."""
>     if end < start or step < 1:
>         raise ValueError("--to must be on or after --from and --step must be at least 1")
>     windows, cursor, index = [], start, 1
>     while cursor <= end:
>         stop = min(cursor + timedelta(days=step - 1), end)
>         windows.append(Window(index, cursor, stop))
>         cursor, index = stop + timedelta(days=1), index + 1
>     return windows
> ```

September 1 to October 26 with a seven-day step gives eight windows, the last ending on October 26. September 1 to September 17 gives three, the last one three days long. The offline tests cover both.

## Send one fast request per window

Each window gets the same prompt with its dates filled in. The prompt asks for a `Headline:` line of at most twelve words, two or three cited sentences, and a plain statement when nothing in range came back. The `web_search` override carries the filters and raises `max_results` from the fast effort level's default of 10 to 20, so a busy week can show more sources than a quiet one.

> **Read WINDOW\_PROMPT and MAX\_RESULTS (timeline.py)**
>
> ```python theme={null}
> WINDOW_PROMPT = (
>     "Report what was published about {topic} between {start} and {end} only. Start with "
>     "one line `Headline: ...` of at most twelve words naming the single most important "
>     "development in that window, then write two or three sentences, each cited with "
>     "bracketed markers like [1][2]. If the search returns nothing dated inside that "
>     "window, write `Headline: Nothing found` and say so plainly."
> )
> MAX_RESULTS = 20  # per fast call; the effort default is 10
> ```

The request is one `client.responses.create` call with `effort="fast"`, the prompt, and the tool override:

> **Read the request call (timeline.py)**
>
> ```python theme={null}
> response, retries = await create_with_retry(
>     client, log, window.label, effort="fast", input=prompt,
>     tools=[{"type": "web_search", "max_results": MAX_RESULTS,
>             "filters": window.filters()}],
> )
> ```

The response carries two output items this recipe reads:

* `search_results` lists every page the search returned: `id`, `url`, `title`, `date`, and `last_updated`.
* `message` holds the text, with citation markers like `[3]` that point at a search result's `id`.

`collect` maps the markers back to their results and counts the distinct URLs with a publication date inside the window. That count is what the chart's bars show.

> **Read collect (timeline.py)**
>
> ```python theme={null}
> MARKER = re.compile(r"\[(?:web:)?(\d+)\]")  # fast writes [3]; medium writes [web:3]
>
>
> def collect(response, window: Window | None = None) -> tuple[str, list, list, int, int]:
>     """Text, every search result, cited results, date counts. [3] and [web:3] are result ids."""
>     text, sources = "", []
>     for item in response.output:
>         if item.type == "search_results":
>             sources.extend(source_record(r) for r in item.results)
>         elif item.type == "message":
>             text += "".join(part.text for part in item.content if part.type == "output_text")
>     by_id = {s["id"]: s for s in sources}
>     cited_ids = sorted({int(m) for m in MARKER.findall(text)} & set(by_id))
>     citations = [by_id[i] for i in cited_ids]
>     undated = sum(1 for s in sources if not s["date"])
>     dated_in_window = 0
>     if window is not None:
>         seen = set()
>         for s in sources:
>             if s["date"] and window.start <= date.fromisoformat(s["date"]) <= window.end:
>                 seen.add(s["url"])
>         dated_in_window = len(seen)
>     return text, sources, citations, dated_in_window, undated
> ```

## Run the windows in parallel

The windows are independent, so they run at the same time through `AsyncPerplexity`, with a semaphore capping how many are in flight. The default of 6 fits the 8 QPS of Tier 2 and above on the [rate limits page](https://docs.perplexity.ai/docs/admin/rate-limits-usage-tiers.md). On Tier 0 or Tier 1, pass `--concurrency 1` or `--concurrency 3`.

The client uses `max_retries=0`, so the library's silent retries are off and a 429 shows up in your terminal. `create_with_retry` retries up to twice and honors `Retry-After`.

> **Read create\_with\_retry and run\_window (timeline.py)**
>
> ```python theme={null}
> async def create_with_retry(client, log, label: str, max_retries: int = 2, **kwargs):
>     """Retry 429 only. The library's own retries are off so each retry is visible."""
>     for attempt in range(max_retries + 1):
>         try:
>             return await client.responses.create(**kwargs), attempt
>         except RateLimitError as err:
>             if attempt == max_retries:
>                 raise
>             wait = float(err.response.headers.get("retry-after") or 2 ** attempt)
>             log(f"    {label}: 429, retrying in {wait:.0f}s")
>             await asyncio.sleep(wait)
>
>
> async def run_window(client, sem, topic: str, window: Window, log) -> WindowResult:
>     prompt = WINDOW_PROMPT.format(topic=topic, start=mdy(window.start), end=mdy(window.end))
>     async with sem:
>         t0 = time.perf_counter()
>         response, retries = await create_with_retry(
>             client, log, window.label, effort="fast", input=prompt,
>             tools=[{"type": "web_search", "max_results": MAX_RESULTS,
>                     "filters": window.filters()}],
>         )
>         latency = time.perf_counter() - t0
>     text, sources, citations, dated, undated = collect(response, window)
>     headline, text = split_headline(text)
>     tokens, cost = usage_numbers(response)
>     return WindowResult(window, text, headline, sources, citations, dated, undated,
>                         latency, tokens, cost, response.effort or "", retries)
> ```

`main` ties it together:

* Schedules all the windows with `asyncio.as_completed`, so each prints its stats line and headline as it finishes.
* Sorts the results, writes a first `timeline.md` with the per-window sections, writes `windows.json` with one record per window, and prints the `Week by week` block.
* Starts the synthesis and the sandbox step together with `asyncio.gather`. The synthesis streams to the terminal; the sandbox result prints when both are done.

If either medium effort request fails, the weeks are already on disk.

> **Read main (timeline.py)**
>
> ```python theme={null}
> async def main(args) -> None:
>     windows = split_windows(date.fromisoformat(args.start), date.fromisoformat(args.stop), args.step)
>     out_dir = Path(args.out) / slugify(args.topic)
>     out_dir.mkdir(parents=True, exist_ok=True)
>     log = lambda line: print(line, file=sys.stderr, flush=True)  # noqa: E731
>     client, sem = AsyncPerplexity(max_retries=0), asyncio.Semaphore(args.concurrency)
>     t0, done, results = time.perf_counter(), 0, []
>     print(f"{len(windows)} windows, up to {args.concurrency} fast requests at a time")
>     for coro in asyncio.as_completed([run_window(client, sem, args.topic, w, log) for w in windows]):
>         r = await coro
>         done += 1
>         results.append(r)
>         print(f"[{done}/{len(windows)}] {r.window.label}  {r.dated_in_window} sources  "
>               f"{r.latency:.1f} s  {r.effort}\n      {r.headline[:90]}")
>     results.sort(key=lambda r: r.window.index)
>     write_markdown(out_dir / "timeline.md", args.topic, results)  # windows first, synthesis later
>     write_windows_json(out_dir / "windows.json", results)
>     print("\nWeek by week\n" + week_by_week(results))
>     print("\nsynthesis (medium, streaming) and sandbox step (medium + sandbox tool) in parallel:\n")
>     synth, cov = await asyncio.gather(run_synthesis(client, args.topic, results, sys.stdout),
>                                       run_coverage(client, args.topic, results))
>     wall = time.perf_counter() - t0
>     print("\n\n" + coverage_summary(cov) + "\n\n" + cov.text)
>     (out_dir / "coverage.py").write_text(cov.code, encoding="utf-8")  # the code the model ran
>     write_markdown(out_dir / "timeline.md", args.topic, results, synth, cov)
>     draw_chart(out_dir / "timeline.png", results, synth.turning_points, args.topic)
>     print("\n" + summary_table(results, synth, cov, wall))
>     print(f"wrote timeline.md, timeline.png, windows.json and coverage.py in {out_dir}")
> ```

## Describe how the story changed

A fast request sees one window, so it cannot tell you that late September's delivery estimates contradict mid-October's order-cut reports. The synthesis step sends all eight window summaries in one `medium` request, because reading them and reconciling them takes more than one model step.

`synthesis_input` lays out every window as a heading, its summary, and the URLs it cited. `SYNTHESIS_PROMPT` asks for three Markdown sections and fixes the format of the turning-point lines so your code can parse them back out for the chart.

> **Read SYNTHESIS\_PROMPT and synthesis\_input (timeline.py)**
>
> ```python theme={null}
> SYNTHESIS_PROMPT = (
>     "Below are week by week summaries of coverage about {topic}, each with the "
>     "sources it cited. Write three parts in Markdown.\n"
>     "## Narrative\nA short narrative of how the coverage changed across the range.\n"
>     "## Turning points\nAt most four windows where the story changed direction, not a line "
>     "per window. Format each exactly as `- YYYY-MM-DD to YYYY-MM-DD: label | detail`, using "
>     "the window dates below, a label of at most six words, and one cited sentence of detail.\n"
>     "## Contradictions\nWindows whose claims conflict with each other, one line each, "
>     "or the single line `None found.`\n"
>     "Cite only the URLs listed below, as Markdown links, or results from searches you "
>     "run yourself, with [n] markers. Do not invent sources.\n\n{windows}"
> )
>
>
> def synthesis_input(topic: str, results: list[WindowResult]) -> str:
>     blocks = []
>     for r in results:
>         urls = "\n".join(f"  - {c['url']}" for c in r.citations) or "  - (no cited sources)"
>         blocks.append(f"### {r.window.start} to {r.window.end}\n{r.text}\nSources:\n{urls}")
>     return SYNTHESIS_PROMPT.format(topic=topic, windows="\n\n".join(blocks))
> ```

The request uses `effort="medium"` and `stream=True`. It also gives the synthesis's own `web_search` the whole range as a date filter, so a search it runs to check a claim does not pull in today's coverage. The loop prints each `response.output_text.delta` as it arrives and reads the final response, with usage and `effort` version, from the `response.completed` event.

> **Read run\_synthesis (timeline.py)**
>
> ```python theme={null}
> async def run_synthesis(client, topic: str, results: list[WindowResult], out) -> Synthesis:
>     """Stream the medium run so the text shows up as it is written. Its own searches get
>     the whole range as a date filter, so it cannot pull in today's coverage."""
>     synth, t0 = Synthesis(), time.perf_counter()
>     whole_range = Window(0, results[0].window.start, results[-1].window.end)
>     stream = await client.responses.create(
>         effort="medium", input=synthesis_input(topic, results), stream=True,
>         tools=[{"type": "web_search", "filters": whole_range.filters()}])
>     async for event in stream:
>         if event.type == "response.output_text.delta":
>             out.write(event.delta)
>             out.flush()
>         elif event.type == "response.completed":
>             final = event.response
>             synth.text, _, synth.citations, _, _ = collect(final)
>             synth.tokens, synth.cost = usage_numbers(final)
>             synth.effort = final.effort or ""
>     synth.latency = time.perf_counter() - t0
>     synth.turning_points = turning_points(synth.text)
>     return synth
> ```

## Count outlets per window with the sandbox tool

The synthesis explains the story. It is not the place for arithmetic: ask a model how many distinct outlets covered each week and you get a plausible number, not a counted one. The [sandbox tool](https://docs.perplexity.ai/docs/agent-api/tools/sandbox.md) fixes that. The model writes Python, a container runs it, and the response carries the code and its output in a `sandbox_results` item.

> [!WARNING]
>
> A request with the sandbox enabled cannot carry `web_search` date filters; the API returns a 400. So the sandbox step is its own request. It does not search; it works on data your code already collected.

Your code reduces each window's results to a list of hostnames and sends that as JSON. The prompt tells the model to save the JSON to a file and parse it, not retype it into code.

> [!TIP]
>
> Without the save-to-file instruction, the model typed all 127 hostnames into its script by hand, wrote the script twice, and used 46,000 tokens. With it, the recorded run's script is 11 lines.

> **Read COVERAGE\_PROMPT and coverage\_input (coverage.py)**
>
> ````python theme={null}
> COVERAGE_PROMPT = (
>     "Below are the outlets (hostnames) behind the search results my code collected about {topic}, "
>     "one list per window, as JSON. In the sandbox, save the JSON exactly as given to data.json, "
>     "then run Python that reads data.json and computes, per window, the number of results and the "
>     "number of distinct outlets, then the outlets that appear in three or more windows with the "
>     "number of windows each appears in. Do not retype the data into code and do not search the "
>     "web. Report only numbers the code printed: a Markdown table with one row per window (window, "
>     "results, distinct outlets), then one line listing the recurring outlets.\n\n```json\n{data}\n```"
> )
>
>
> def coverage_input(topic: str, results: list) -> str:
>     """Hostnames only: the smaller the payload, the less the model has to copy into the file."""
>     data = [{"window": f"{r.window.start} to {r.window.end}",
>              "hosts": [host(s["url"]) for s in r.sources]} for r in results]
>     return COVERAGE_PROMPT.format(topic=topic, data=json.dumps(data, separators=(",", ":")))
> ````

`run_coverage` adds `{"type": "sandbox"}` to the medium effort level's tools, then reads the code and exit code from the `sandbox_results` item, the table from the `message` item, and the session cost from `usage.tool_calls_details`. Your own code could build this table in ten lines. The step is here to show the pattern (send data in, get code and a result back) on numbers you can check against `windows.json`.

> **Read run\_coverage (coverage.py)**
>
> ```python theme={null}
> async def run_coverage(client, topic: str, results: list) -> Coverage:
>     """Medium effort with the sandbox tool merged in. A request that enables the sandbox cannot
>     carry web_search date filters, so this step gets the collected data instead of a search."""
>     cov, t0 = Coverage(), time.perf_counter()
>     response = await client.responses.create(
>         effort="medium", input=coverage_input(topic, results), tools=[{"type": "sandbox"}])
>     cov.latency = time.perf_counter() - t0
>     for item in response.output:
>         if item.type == "sandbox_results":
>             cov.code += item.code
>             for run in item.results:
>                 cov.stdout += run.stdout
>                 cov.exit_code = run.exit_code
>                 cov.duration_ms += run.duration_ms
>         elif item.type == "message":
>             cov.text += "".join(p.text for p in item.content if p.type == "output_text")
>     usage = response.usage
>     if usage is not None:
>         cov.tokens = usage.total_tokens
>         cov.cost = usage.cost.total_cost if usage.cost else None
>         sandbox = (usage.tool_calls_details or {}).get("sandbox")
>         cov.sandbox_cost = sandbox.cost_usd if sandbox else None
>     cov.effort = response.effort or ""
>     return cov
> ```

## Write the timeline and draw the chart

`report.py` makes no API calls. `write_markdown` writes one section per window (headline, the model's text, the sources it cited), then the synthesis, then the sandbox table. `write_windows_json` writes the same data as records.

`turning_points` reads only the text between the "Turning points" heading and the next heading. An earlier version read the whole text, and the Contradictions section, which uses the same `- date to date:` line shape, showed up on the chart as extra markers.

`draw_chart` keeps the attribution straight: the bars come from counts your code took from the search results, the markers from turning points the model named. A turning point gets a marker when its start date matches a window start. When the synthesis merges two windows into one span, the marker sits on the span's first window.

> **Read turning\_points and draw\_chart (report.py)**
>
> ```python theme={null}
> def turning_points(text: str) -> list:
>     """Read only the Turning points section; Contradictions uses the same line shape."""
>     section = text.split("## Turning points", 1)[-1].split("\n## ", 1)[0]
>     return [(date.fromisoformat(a), date.fromisoformat(b), label.strip(" *"))
>             for a, b, label in TURNING_POINT.findall(section)]
>
>
> def draw_chart(path: Path, results: list, turning_points: list, topic: str):
>     """Bars are counts the search returned; numbered markers are turning points the model named."""
>     import matplotlib
>     matplotlib.use("Agg")
>     import matplotlib.pyplot as plt
>     from matplotlib.ticker import MaxNLocator
>
>     starts = [r.window.start for r in results]
>     counts = [r.dated_in_window for r in results]
>     labels = [f"{r.window.start:%b %d}" for r in results]
>     xs = list(range(len(results)))
>     fig, ax = plt.subplots(figsize=(10, 5.2))
>     ax.bar(xs, counts, color="#8fb3c9", width=0.7)
>     notes = []
>     for tp_start, _, label in turning_points:
>         if tp_start in starts:
>             i = starts.index(tp_start)
>             notes.append(f"{len(notes) + 1}. {tp_start:%b %d}: {label}")
>             ax.plot(i, counts[i] + 0.6, "v", color="#c0392b", markersize=10)
>             ax.annotate(str(len(notes)), (i, counts[i] + 0.6), xytext=(0, 9),
>                         textcoords="offset points", ha="center", fontsize=10, weight="bold")
>     ax.set_xticks(xs, labels)
>     ax.yaxis.set_major_locator(MaxNLocator(integer=True))
>     ax.set_xlabel("window start")
>     ax.set_ylabel("distinct dated sources in window")
>     ax.set_title(f"{topic}: sources per window and turning points")
>     ax.margins(y=0.2)
>     if notes:
>         fig.text(0.02, 0.01, "\n".join(notes), fontsize=9, va="bottom")
>         fig.subplots_adjust(bottom=0.12 + 0.045 * len(notes))
>     fig.savefig(path, dpi=150)
>     plt.close(fig)
> ```

## Save the full files

Save these files next to `requirements.txt`. `timeline.py` imports from `report.py` and `coverage.py`; `compare.py` imports from `timeline.py`.

> **Read timeline.py (full file, 241 lines)**
>
> ```python timeline.py theme={null}
> """Track what the web said each week: python timeline.py "iPhone 18 Pro" --from 2026-09-01 --to 2026-10-26"""
>
> import argparse
> import asyncio
> import re
> import sys
> import time
> from dataclasses import dataclass, field
> from datetime import date, timedelta
> from pathlib import Path
>
> from perplexity import AsyncPerplexity, RateLimitError
>
> from coverage import coverage_summary, run_coverage
> from report import (draw_chart, slugify, split_headline, summary_table, turning_points,
>                     week_by_week, write_markdown, write_windows_json)
>
> WINDOW_PROMPT = (
>     "Report what was published about {topic} between {start} and {end} only. Start with "
>     "one line `Headline: ...` of at most twelve words naming the single most important "
>     "development in that window, then write two or three sentences, each cited with "
>     "bracketed markers like [1][2]. If the search returns nothing dated inside that "
>     "window, write `Headline: Nothing found` and say so plainly."
> )
>
> SYNTHESIS_PROMPT = (
>     "Below are week by week summaries of coverage about {topic}, each with the "
>     "sources it cited. Write three parts in Markdown.\n"
>     "## Narrative\nA short narrative of how the coverage changed across the range.\n"
>     "## Turning points\nAt most four windows where the story changed direction, not a line "
>     "per window. Format each exactly as `- YYYY-MM-DD to YYYY-MM-DD: label | detail`, using "
>     "the window dates below, a label of at most six words, and one cited sentence of detail.\n"
>     "## Contradictions\nWindows whose claims conflict with each other, one line each, "
>     "or the single line `None found.`\n"
>     "Cite only the URLs listed below, as Markdown links, or results from searches you "
>     "run yourself, with [n] markers. Do not invent sources.\n\n{windows}"
> )
>
> MARKER = re.compile(r"\[(?:web:)?(\d+)\]")  # fast writes [3]; medium writes [web:3]
> MAX_RESULTS = 20  # per fast call; the effort default is 10
>
>
> @dataclass
> class Window:
>     index: int
>     start: date
>     end: date
>
>     @property
>     def label(self) -> str:
>         return f"{self.start:%b %d} to {self.end:%b %d}"
>
>     def filters(self) -> dict:
>         return {
>             "search_after_date_filter": mdy(self.start),
>             "search_before_date_filter": mdy(self.end),
>         }
>
>
> @dataclass
> class WindowResult:
>     window: Window
>     text: str = ""
>     headline: str = ""
>     sources: list = field(default_factory=list)  # every search result, in id order
>     citations: list = field(default_factory=list)  # the ones the text cites
>     dated_in_window: int = 0
>     undated: int = 0
>     latency: float = 0.0
>     tokens: int = 0
>     cost: float | None = None
>     effort: str = ""
>     retries: int = 0
>
>
> @dataclass
> class Synthesis:
>     text: str = ""
>     citations: list = field(default_factory=list)  # results its own searches returned and it cited
>     turning_points: list = field(default_factory=list)  # (start, end, label)
>     latency: float = 0.0
>     tokens: int = 0
>     cost: float | None = None
>     effort: str = ""
>
>
> def mdy(d: date) -> str:  # date filters take MM/DD/YYYY, not ISO
>     return f"{d.month:02d}/{d.day:02d}/{d.year}"
>
>
> def split_windows(start: date, end: date, step: int) -> list[Window]:
>     """Inclusive windows of `step` days; the last one is as short as it needs to be."""
>     if end < start or step < 1:
>         raise ValueError("--to must be on or after --from and --step must be at least 1")
>     windows, cursor, index = [], start, 1
>     while cursor <= end:
>         stop = min(cursor + timedelta(days=step - 1), end)
>         windows.append(Window(index, cursor, stop))
>         cursor, index = stop + timedelta(days=1), index + 1
>     return windows
>
>
> def source_record(result) -> dict:
>     return {
>         "id": result.id,
>         "url": result.url,
>         "title": result.title,
>         "date": result.date,
>         "last_updated": result.last_updated,
>     }
>
>
> def collect(response, window: Window | None = None) -> tuple[str, list, list, int, int]:
>     """Text, every search result, cited results, date counts. [3] and [web:3] are result ids."""
>     text, sources = "", []
>     for item in response.output:
>         if item.type == "search_results":
>             sources.extend(source_record(r) for r in item.results)
>         elif item.type == "message":
>             text += "".join(part.text for part in item.content if part.type == "output_text")
>     by_id = {s["id"]: s for s in sources}
>     cited_ids = sorted({int(m) for m in MARKER.findall(text)} & set(by_id))
>     citations = [by_id[i] for i in cited_ids]
>     undated = sum(1 for s in sources if not s["date"])
>     dated_in_window = 0
>     if window is not None:
>         seen = set()
>         for s in sources:
>             if s["date"] and window.start <= date.fromisoformat(s["date"]) <= window.end:
>                 seen.add(s["url"])
>         dated_in_window = len(seen)
>     return text, sources, citations, dated_in_window, undated
>
>
> def usage_numbers(response) -> tuple[int, float | None]:
>     usage = response.usage
>     if usage is None:
>         return 0, None
>     cost = usage.cost.total_cost if usage.cost else None
>     return usage.total_tokens, cost
>
>
> async def create_with_retry(client, log, label: str, max_retries: int = 2, **kwargs):
>     """Retry 429 only. The library's own retries are off so each retry is visible."""
>     for attempt in range(max_retries + 1):
>         try:
>             return await client.responses.create(**kwargs), attempt
>         except RateLimitError as err:
>             if attempt == max_retries:
>                 raise
>             wait = float(err.response.headers.get("retry-after") or 2 ** attempt)
>             log(f"    {label}: 429, retrying in {wait:.0f}s")
>             await asyncio.sleep(wait)
>
>
> async def run_window(client, sem, topic: str, window: Window, log) -> WindowResult:
>     prompt = WINDOW_PROMPT.format(topic=topic, start=mdy(window.start), end=mdy(window.end))
>     async with sem:
>         t0 = time.perf_counter()
>         response, retries = await create_with_retry(
>             client, log, window.label, effort="fast", input=prompt,
>             tools=[{"type": "web_search", "max_results": MAX_RESULTS,
>                     "filters": window.filters()}],
>         )
>         latency = time.perf_counter() - t0
>     text, sources, citations, dated, undated = collect(response, window)
>     headline, text = split_headline(text)
>     tokens, cost = usage_numbers(response)
>     return WindowResult(window, text, headline, sources, citations, dated, undated,
>                         latency, tokens, cost, response.effort or "", retries)
>
>
> def synthesis_input(topic: str, results: list[WindowResult]) -> str:
>     blocks = []
>     for r in results:
>         urls = "\n".join(f"  - {c['url']}" for c in r.citations) or "  - (no cited sources)"
>         blocks.append(f"### {r.window.start} to {r.window.end}\n{r.text}\nSources:\n{urls}")
>     return SYNTHESIS_PROMPT.format(topic=topic, windows="\n\n".join(blocks))
>
>
> async def run_synthesis(client, topic: str, results: list[WindowResult], out) -> Synthesis:
>     """Stream the medium run so the text shows up as it is written. Its own searches get
>     the whole range as a date filter, so it cannot pull in today's coverage."""
>     synth, t0 = Synthesis(), time.perf_counter()
>     whole_range = Window(0, results[0].window.start, results[-1].window.end)
>     stream = await client.responses.create(
>         effort="medium", input=synthesis_input(topic, results), stream=True,
>         tools=[{"type": "web_search", "filters": whole_range.filters()}])
>     async for event in stream:
>         if event.type == "response.output_text.delta":
>             out.write(event.delta)
>             out.flush()
>         elif event.type == "response.completed":
>             final = event.response
>             synth.text, _, synth.citations, _, _ = collect(final)
>             synth.tokens, synth.cost = usage_numbers(final)
>             synth.effort = final.effort or ""
>     synth.latency = time.perf_counter() - t0
>     synth.turning_points = turning_points(synth.text)
>     return synth
>
>
> async def main(args) -> None:
>     windows = split_windows(date.fromisoformat(args.start), date.fromisoformat(args.stop), args.step)
>     out_dir = Path(args.out) / slugify(args.topic)
>     out_dir.mkdir(parents=True, exist_ok=True)
>     log = lambda line: print(line, file=sys.stderr, flush=True)  # noqa: E731
>     client, sem = AsyncPerplexity(max_retries=0), asyncio.Semaphore(args.concurrency)
>     t0, done, results = time.perf_counter(), 0, []
>     print(f"{len(windows)} windows, up to {args.concurrency} fast requests at a time")
>     for coro in asyncio.as_completed([run_window(client, sem, args.topic, w, log) for w in windows]):
>         r = await coro
>         done += 1
>         results.append(r)
>         print(f"[{done}/{len(windows)}] {r.window.label}  {r.dated_in_window} sources  "
>               f"{r.latency:.1f} s  {r.effort}\n      {r.headline[:90]}")
>     results.sort(key=lambda r: r.window.index)
>     write_markdown(out_dir / "timeline.md", args.topic, results)  # windows first, synthesis later
>     write_windows_json(out_dir / "windows.json", results)
>     print("\nWeek by week\n" + week_by_week(results))
>     print("\nsynthesis (medium, streaming) and sandbox step (medium + sandbox tool) in parallel:\n")
>     synth, cov = await asyncio.gather(run_synthesis(client, args.topic, results, sys.stdout),
>                                       run_coverage(client, args.topic, results))
>     wall = time.perf_counter() - t0
>     print("\n\n" + coverage_summary(cov) + "\n\n" + cov.text)
>     (out_dir / "coverage.py").write_text(cov.code, encoding="utf-8")  # the code the model ran
>     write_markdown(out_dir / "timeline.md", args.topic, results, synth, cov)
>     draw_chart(out_dir / "timeline.png", results, synth.turning_points, args.topic)
>     print("\n" + summary_table(results, synth, cov, wall))
>     print(f"wrote timeline.md, timeline.png, windows.json and coverage.py in {out_dir}")
>
>
> if __name__ == "__main__":
>     p = argparse.ArgumentParser(description=__doc__)
>     p.add_argument("topic")
>     p.add_argument("--from", dest="start", required=True, help="YYYY-MM-DD")
>     p.add_argument("--to", dest="stop", required=True, help="YYYY-MM-DD")
>     p.add_argument("--step", type=int, default=7, help="days per window")
>     p.add_argument("--concurrency", type=int, default=6)
>     p.add_argument("--out", default="out")
>     asyncio.run(main(p.parse_args()))
> ```

> **Read report.py (full file, 131 lines)**
>
> ```python report.py theme={null}
> """Write timeline.md, windows.json and timeline.png from the window results and the synthesis.
>
> The bars come from counts the search returned; the markers come from turning points the
> medium run named. Nothing here calls the API."""
>
> import json
> import re
> import statistics
> from datetime import date
> from pathlib import Path
>
> MARKER = re.compile(r"\s*\[(?:web:)?\d+\]")
> TURNING_POINT = re.compile(
>     r"^-\s*(\d{4}-\d{2}-\d{2}) to (\d{4}-\d{2}-\d{2}):\s*([^|\n]+?)\s*(?:\|.*)?$", re.M)
>
>
> def turning_points(text: str) -> list:
>     """Read only the Turning points section; Contradictions uses the same line shape."""
>     section = text.split("## Turning points", 1)[-1].split("\n## ", 1)[0]
>     return [(date.fromisoformat(a), date.fromisoformat(b), label.strip(" *"))
>             for a, b, label in TURNING_POINT.findall(section)]
>
>
> def split_headline(text: str) -> tuple[str, str]:
>     """Return (headline, body). The fast prompt asks for a `Headline:` first line; if the
>     model skipped it, fall back to the first sentence."""
>     stripped = text.strip()
>     m = re.match(r"\**\s*Headline:\**\s*(.+?)\**\s*(?:\n+|$)", stripped)
>     if m:
>         headline, body = m.group(1), stripped[m.end():].strip()
>     else:
>         headline, body = re.split(r"(?<=[.!?])\s", stripped, 1)[0], stripped
>     return MARKER.sub("", headline).strip().rstrip("."), body
>
>
> def week_by_week(results: list) -> str:
>     return "\n".join(f"{r.window.label}  {r.headline[:90]}" for r in results)
>
>
> def write_windows_json(path: Path, results: list):
>     """One record per window so you can build on the data without re-running."""
>     records = [{
>         "index": r.window.index, "start": str(r.window.start), "end": str(r.window.end),
>         "headline": r.headline, "summary": r.text, "sources": r.citations, "results": r.sources,
>         "dated_in_window": r.dated_in_window, "undated": r.undated,
>         "latency_s": round(r.latency, 2), "tokens": r.tokens, "cost": r.cost, "effort": r.effort,
>     } for r in results]
>     path.write_text(json.dumps(records, indent=2), encoding="utf-8")
>
>
> def write_markdown(path: Path, topic: str, results: list, synth=None, cov=None):
>     """Write the per-window sections; add the synthesis and sandbox numbers when present."""
>     lines = [f"# {topic}: {results[0].window.start} to {results[-1].window.end}", ""]
>     for r in results:
>         lines += [f"## {r.window.label} ({r.dated_in_window} dated sources)", "",
>                   f"**{r.headline}**", "", r.text, ""]
>         for c in r.citations:
>             lines.append(f"- [{c['id']}] {c['date'] or 'undated'} [{c['title']}]({c['url']})")
>         lines.append("")
>     if synth is None:
>         path.write_text("\n".join(lines) + "\n", encoding="utf-8")
>         return
>     lines += ["# How the story turned", "", synth.text, ""]
>     if synth.citations:
>         lines += ["Sources the synthesis run found and cited itself:", ""]
>         lines += [f"- [{s['id']}] {s['date'] or 'undated'} [{s['title']}]({s['url']})"
>                   for s in synth.citations]
>     if cov is not None and cov.text:
>         lines += ["", "# Coverage numbers, computed in the sandbox", "", cov.text, "",
>                   "The Python the model ran is saved next to this file as coverage.py."]
>     path.write_text("\n".join(lines) + "\n", encoding="utf-8")
>
>
> def draw_chart(path: Path, results: list, turning_points: list, topic: str):
>     """Bars are counts the search returned; numbered markers are turning points the model named."""
>     import matplotlib
>     matplotlib.use("Agg")
>     import matplotlib.pyplot as plt
>     from matplotlib.ticker import MaxNLocator
>
>     starts = [r.window.start for r in results]
>     counts = [r.dated_in_window for r in results]
>     labels = [f"{r.window.start:%b %d}" for r in results]
>     xs = list(range(len(results)))
>     fig, ax = plt.subplots(figsize=(10, 5.2))
>     ax.bar(xs, counts, color="#8fb3c9", width=0.7)
>     notes = []
>     for tp_start, _, label in turning_points:
>         if tp_start in starts:
>             i = starts.index(tp_start)
>             notes.append(f"{len(notes) + 1}. {tp_start:%b %d}: {label}")
>             ax.plot(i, counts[i] + 0.6, "v", color="#c0392b", markersize=10)
>             ax.annotate(str(len(notes)), (i, counts[i] + 0.6), xytext=(0, 9),
>                         textcoords="offset points", ha="center", fontsize=10, weight="bold")
>     ax.set_xticks(xs, labels)
>     ax.yaxis.set_major_locator(MaxNLocator(integer=True))
>     ax.set_xlabel("window start")
>     ax.set_ylabel("distinct dated sources in window")
>     ax.set_title(f"{topic}: sources per window and turning points")
>     ax.margins(y=0.2)
>     if notes:
>         fig.text(0.02, 0.01, "\n".join(notes), fontsize=9, va="bottom")
>         fig.subplots_adjust(bottom=0.12 + 0.045 * len(notes))
>     fig.savefig(path, dpi=150)
>     plt.close(fig)
>
>
> def summary_table(results: list, synth, cov, wall: float) -> str:
>     lat = sorted(r.latency for r in results)
>     p50 = statistics.median(lat)
>     p95 = lat[min(len(lat) - 1, round(0.95 * (len(lat) - 1)))]
>     costs = [r.cost for r in results if r.cost is not None]
>     fast_cost = f"${sum(costs):.4f}" if costs else "n/a"
>     rows = [
>         ("windows", str(len(results))),
>         ("fast calls (incl. retries)", str(len(results) + sum(r.retries for r in results))),
>         ("fast latency p50 / p95", f"{p50:.1f} s / {p95:.1f} s"),
>         ("fast tokens / cost", f"{sum(r.tokens for r in results):,} / {fast_cost}"),
>         ("medium latency / tokens / cost", f"{synth.latency:.1f} s / {synth.tokens:,} / "
>          + (f"${synth.cost:.4f}" if synth.cost is not None else "n/a")),
>         ("sandbox step latency / tokens / cost", f"{cov.latency:.1f} s / {cov.tokens:,} / "
>          + (f"${cov.cost:.4f}" if cov.cost is not None else "n/a")),
>         ("effort versions", f"{results[0].effort}, {synth.effort}"),
>         ("wall time", f"{wall:.1f} s"),
>     ]
>     width = max(len(k) for k, _ in rows)
>     return "\n".join(f"{k:<{width}}  {v}" for k, v in rows)
>
>
> def slugify(topic: str) -> str:
>     return re.sub(r"[^a-z0-9]+", "-", topic.lower()).strip("-")
> ```

> **Read coverage.py (full file, 76 lines)**
>
> ````python coverage.py theme={null}
> """Third request: the model computes cross-window numbers in the sandbox from the search
> results your code collected. The code it ran comes back in the response and is saved."""
>
> import json
> import time
> from dataclasses import dataclass
> from urllib.parse import urlparse
>
> COVERAGE_PROMPT = (
>     "Below are the outlets (hostnames) behind the search results my code collected about {topic}, "
>     "one list per window, as JSON. In the sandbox, save the JSON exactly as given to data.json, "
>     "then run Python that reads data.json and computes, per window, the number of results and the "
>     "number of distinct outlets, then the outlets that appear in three or more windows with the "
>     "number of windows each appears in. Do not retype the data into code and do not search the "
>     "web. Report only numbers the code printed: a Markdown table with one row per window (window, "
>     "results, distinct outlets), then one line listing the recurring outlets.\n\n```json\n{data}\n```"
> )
>
>
> @dataclass
> class Coverage:
>     text: str = ""
>     code: str = ""
>     stdout: str = ""
>     exit_code: int | None = None
>     duration_ms: int = 0
>     latency: float = 0.0
>     tokens: int = 0
>     cost: float | None = None
>     sandbox_cost: float | None = None
>     effort: str = ""
>
>
> def host(url: str) -> str:
>     return (urlparse(url).hostname or "").removeprefix("www.")
>
>
> def coverage_input(topic: str, results: list) -> str:
>     """Hostnames only: the smaller the payload, the less the model has to copy into the file."""
>     data = [{"window": f"{r.window.start} to {r.window.end}",
>              "hosts": [host(s["url"]) for s in r.sources]} for r in results]
>     return COVERAGE_PROMPT.format(topic=topic, data=json.dumps(data, separators=(",", ":")))
>
>
> async def run_coverage(client, topic: str, results: list) -> Coverage:
>     """Medium effort with the sandbox tool merged in. A request that enables the sandbox cannot
>     carry web_search date filters, so this step gets the collected data instead of a search."""
>     cov, t0 = Coverage(), time.perf_counter()
>     response = await client.responses.create(
>         effort="medium", input=coverage_input(topic, results), tools=[{"type": "sandbox"}])
>     cov.latency = time.perf_counter() - t0
>     for item in response.output:
>         if item.type == "sandbox_results":
>             cov.code += item.code
>             for run in item.results:
>                 cov.stdout += run.stdout
>                 cov.exit_code = run.exit_code
>                 cov.duration_ms += run.duration_ms
>         elif item.type == "message":
>             cov.text += "".join(p.text for p in item.content if p.type == "output_text")
>     usage = response.usage
>     if usage is not None:
>         cov.tokens = usage.total_tokens
>         cov.cost = usage.cost.total_cost if usage.cost else None
>         sandbox = (usage.tool_calls_details or {}).get("sandbox")
>         cov.sandbox_cost = sandbox.cost_usd if sandbox else None
>     cov.effort = response.effort or ""
>     return cov
>
>
> def coverage_summary(cov: Coverage) -> str:
>     cost = f"${cov.cost:.4f}" if cov.cost is not None else "n/a"
>     session = f" incl. ${cov.sandbox_cost:.2f} sandbox session" if cov.sandbox_cost else ""
>     return (f"sandbox step ({cov.effort} + sandbox tool): {len(cov.code.splitlines())} lines of "
>             f"Python, exit {cov.exit_code}, {cov.duration_ms} ms in the container, "
>             f"{cov.latency:.1f} s request, {cov.tokens:,} tokens, {cost}{session}")
> ````

> **Read compare.py (full file, 52 lines)**
>
> ```python compare.py theme={null}
> """Run one window at several effort levels and print what changed.
>
> python compare.py "iPhone 18 Pro" --from 2026-09-22 --to 2026-09-28 --effort fast medium high
> """
>
> import argparse
> import asyncio
> import time
> from datetime import date
>
> from perplexity import AsyncPerplexity
>
> from timeline import MAX_RESULTS, WINDOW_PROMPT, Window, collect, mdy, usage_numbers
>
> TOOL_ITEMS = {"search_results", "fetch_url_results"}
>
>
> async def one(client, effort: str, topic: str, window: Window) -> dict:
>     prompt = WINDOW_PROMPT.format(topic=topic, start=mdy(window.start), end=mdy(window.end))
>     t0 = time.perf_counter()
>     response = await client.responses.create(
>         effort=effort, input=prompt,
>         tools=[{"type": "web_search", "max_results": MAX_RESULTS, "filters": window.filters()}])
>     latency = time.perf_counter() - t0
>     text, sources, citations, dated, _ = collect(response, window)
>     tokens, cost = usage_numbers(response)
>     tool_items = sum(1 for item in response.output if item.type in TOOL_ITEMS)
>     return {"effort": response.effort, "latency": latency, "tokens": tokens, "cost": cost,
>             "tool_items": tool_items, "dated": dated, "cited": len(citations), "text": text}
>
>
> async def main(args):
>     window = Window(1, date.fromisoformat(args.start), date.fromisoformat(args.stop))
>     client = AsyncPerplexity(max_retries=0)
>     runs = await asyncio.gather(*[one(client, e, args.topic, window) for e in args.effort])
>     header = f"{'':<22}" + "".join(f"{r['effort']:<24}" for r in runs)
>     print(header)
>     for key, fmt in [("latency", "{:.1f} s"), ("tokens", "{:,}"), ("cost", "${:.4f}"),
>                      ("tool_items", "{}"), ("dated", "{}"), ("cited", "{}")]:
>         cells = [fmt.format(r[key]) if r[key] is not None else "n/a" for r in runs]
>         print(f"{key:<22}" + "".join(f"{c:<24}" for c in cells))
>     for r in runs:
>         print(f"\n--- {r['effort']} ---\n{r['text']}")
>
>
> if __name__ == "__main__":
>     p = argparse.ArgumentParser(description=__doc__)
>     p.add_argument("topic")
>     p.add_argument("--from", dest="start", required=True)
>     p.add_argument("--to", dest="stop", required=True)
>     p.add_argument("--effort", nargs="+", default=["fast", "medium", "high"])
>     asyncio.run(main(p.parse_args()))
> ```

## Run the script

```bash theme={null}
export PERPLEXITY_API_KEY=pplx-...
python timeline.py "iPhone 18 Pro" --from 2026-09-01 --to 2026-10-26 --step 7
```

The iPhone 18 Pro, from the week before Apple's September 9 event to October 26, is a dense story with primary sources you can check. This is the recorded run's terminal output, trimmed: the streamed turning points, the sandbox stats line, the summary table, and the per-request timings are cut, and the `--out` path is shown as the default `out/`:

```text theme={null}
8 windows, up to 6 fast requests at a time
[1/8] Oct 13 to Oct 19  19 sources  fast@2026-10-16.0
      Reports diverged on iPhone 18 Pro demand
[2/8] Sep 08 to Sep 14  14 sources  fast@2026-10-16.0
      Apple unveiled iPhone 18 Pro with variable-aperture camera
[3/8] Oct 06 to Oct 12  15 sources  fast@2026-10-16.0
      Apple reportedly cut iPhone 18 Pro component orders
[4/8] Sep 29 to Oct 05  17 sources  fast@2026-10-16.0
      Variable-aperture camera reaches iPhone 18 Pro video
[5/8] Sep 22 to Sep 28  15 sources  fast@2026-10-16.0
      Face ID glitches reported on some iPhone 18 Pro models
[6/8] Sep 01 to Sep 07  13 sources  fast@2026-10-16.0
      iPhone 18 Pro expected to debut September 9
[7/8] Sep 15 to Sep 21  13 sources  fast@2026-10-16.0
      iPhone 18 Pro launched with upgraded camera and chip
[8/8] Oct 20 to Oct 26  20 sources  fast@2026-10-16.0
      Apple expanded U.S. manufacturing partnerships for iPhone 18 Pro

Week by week
Sep 01 to Sep 07  iPhone 18 Pro expected to debut September 9
Sep 08 to Sep 14  Apple unveiled iPhone 18 Pro with variable-aperture camera
Sep 15 to Sep 21  iPhone 18 Pro launched with upgraded camera and chip
Sep 22 to Sep 28  Face ID glitches reported on some iPhone 18 Pro models
Sep 29 to Oct 05  Variable-aperture camera reaches iPhone 18 Pro video
Oct 06 to Oct 12  Apple reportedly cut iPhone 18 Pro component orders
Oct 13 to Oct 19  Reports diverged on iPhone 18 Pro demand
Oct 20 to Oct 26  Apple expanded U.S. manufacturing partnerships for iPhone 18 Pro

synthesis (medium, streaming) and sandbox step (medium + sandbox tool) in parallel:

## Narrative

Coverage moved from pre-event speculation about the iPhone 18 Pro’s design, features, and price to confirmation of its camera and launch details. After release, attention shifted to early bugs and hardware complaints, then to conflicting readings of demand. By late October, stories emphasized U.S. suppliers and customer repair options. [1](https://www.apple.com/newsroom/2026/09/apple-debuts-iphone-18-pro-and-iphone-18-pro-max/) [2](https://www.theapplepost.com/2026/09/22/72611/some-iphone-18-pro-users-report-face-id-problems-after-launch/) [3](https://www.macrumors.com/2026/09/29/iphone-18-pro-speaker-crackling/) [4](https://finance.yahoo.com/technology/ai/articles/apple-cuts-iphone-18-pro-181700565.html) [5](https://www.macrumors.com/2026/10/20/apple-touts-iphone-duo-18-pro-parts-made-in-usa/)

[turning points and contradictions streamed here; see timeline.md below]

[sandbox step stats line trimmed; the code it ran is saved as coverage.py]

| window | results | distinct outlets |
|---|---:|---:|
| 2026-09-01 to 2026-09-07 | 13 | 7 |
| 2026-09-08 to 2026-09-14 | 14 | 10 |
| 2026-09-15 to 2026-09-21 | 13 | 7 |
| 2026-09-22 to 2026-09-28 | 15 | 14 |
| 2026-09-29 to 2026-10-05 | 17 | 14 |
| 2026-10-06 to 2026-10-12 | 15 | 13 |
| 2026-10-13 to 2026-10-19 | 19 | 19 |
| 2026-10-20 to 2026-10-26 | 20 | 15 |

Recurring outlets: 9to5mac.com (4), apple.com (3), cnet.com (3), cultofmac.com (3),
finance.yahoo.com (3), gsmarena.com (3), macdailynews.com (3), macrumors.com (7)
[summary table trimmed]
wrote timeline.md, timeline.png, windows.json and coverage.py in out/iphone-18-pro
```

Keep the effort version strings. `fast@2026-10-16.0` names the exact model, prompt, tools, and limits that produced these results. Pass that string as `effort` instead of plain `fast` to make tomorrow's run behave like today's.

> **Read timeline.md from the recorded run (trimmed to three windows, the synthesis, and the sandbox table)**
>
> ```markdown theme={null}
> # iPhone 18 Pro: 2026-09-01 to 2026-10-26
>
> ## Sep 08 to Sep 14 (14 dated sources)
>
> **Apple unveiled iPhone 18 Pro with variable-aperture camera**
>
> Apple announced the iPhone 18 Pro on September 9, highlighting a 48-megapixel main camera with
> variable aperture and new Pro camera controls[1][2]. Reports also covered preorders, which began
> September 12; by September 14, delivery estimates for some configurations had slipped beyond the
> September 18 launch date[3][10].
>
> - [1] 2026-09-09 [Apple debuts iPhone 18 Pro and iPhone 18 Pro Max - Apple](https://www.apple.com/newsroom/2026/09/apple-debuts-iphone-18-pro-and-iphone-18-pro-max/)
> - [2] 2026-09-09 [Apple launches iPhone 18 Pro with upgraded camera | TechCrunch](https://techcrunch.com/2026/09/09/apple-launches-iphone-18-pro-with-upgraded-camera/)
> - [3] 2026-09-09 [Apple Announces iPhone 18 Pro and iPhone 18 Pro Max](https://www.macrumors.com/2026/09/09/apple-announces-iphone-18-pro-and-iphone-18-pro-max/)
> - [10] 2026-09-14 [Didn't Pre-Order an iPhone 18 Pro Yet? Wait Times Are Even ...](https://www.macrumors.com/2026/09/14/iphone-18-pro-pre-orders-shipping-delivery-dates/)
>
> ## Sep 22 to Sep 28 (15 dated sources)
>
> **Face ID glitches reported on some iPhone 18 Pro models**
>
> Some owners reported Face ID failures, freezing, and setup errors shortly after launch; the reports
> were limited, and no cause had been identified by September 22.[7] A September 26 report also
> described a Face ID reboot bug and reduced storage speed under sustained heavy workloads on 1TB Pro
> Max models, with Apple said to be preparing a software fix for the Face ID issue.[5]
>
> - [5] 2026-09-26 [iPhone 18 Pro's got some '-gates' [Cult of Mac podcast 34]](https://www.cultofmac.com/news/iphone-18-pro-problems-cult-of-mac-podcast-34)
> - [7] 2026-09-22 [Some iPhone 18 Pro users report Face ID problems after launch](https://www.theapplepost.com/2026/09/22/72611/some-iphone-18-pro-users-report-face-id-problems-after-launch/)
>
> ## Oct 06 to Oct 12 (15 dated sources)
>
> **Apple reportedly cut iPhone 18 Pro component orders**
>
> Reports published October 9–12 said Apple asked some suppliers to reduce October component orders
> for the iPhone 18 Pro models by at least 15%, citing softer demand and higher memory-chip costs;
> Apple had not confirmed the cuts.[1][3] On October 12, a separate report said JPMorgan viewed the
> adjustment as typical and reported stronger demand for the Pro models in China than for the prior
> year’s models.[5]
>
> - [1] 2026-10-12 [Apple cuts iPhone 18 Pro orders as AI raises the bill - Yahoo Finance](https://finance.yahoo.com/technology/ai/articles/apple-cuts-iphone-18-pro-181700565.html)
> - [3] 2026-10-09 [Apple cuts iPhone 18 Pro orders due to soft demand: Report](https://indianexpress.com/article/technology/apple-cuts-iphone-18-pro-orders-due-to-soft-demand-report-10914405/)
> - [5] 2026-10-12 [JP Morgan: demand for iPhone 18 Pro models is stronger than for the 17 Pro models last year](https://www.gsmarena.com/jp_morgan_demand_for_iphone_18_pro_models_is_stronger_than_for_the_17_pro_models_last_year-news-75005.php)
>
> [Sep 01, Sep 15, Sep 29, Oct 13 and Oct 20 sections trimmed]
>
> # How the story turned
>
> ## Narrative
>
> Coverage moved from pre-event speculation about the iPhone 18 Pro’s design, features, and price to confirmation of its camera and launch details. After release, attention shifted to early bugs and hardware complaints, then to conflicting readings of demand. By late October, stories emphasized U.S. suppliers and customer repair options. [1](https://www.apple.com/newsroom/2026/09/apple-debuts-iphone-18-pro-and-iphone-18-pro-max/) [2](https://www.theapplepost.com/2026/09/22/72611/some-iphone-18-pro-users-report-face-id-problems-after-launch/) [3](https://www.macrumors.com/2026/09/29/iphone-18-pro-speaker-crackling/) [4](https://finance.yahoo.com/technology/ai/articles/apple-cuts-iphone-18-pro-181700565.html) [5](https://www.macrumors.com/2026/10/20/apple-touts-iphone-duo-18-pro-parts-made-in-usa/)
>
> ## Turning points
>
> - 2026-09-08 to 2026-09-14: Rumor becomes product | Apple announced the iPhone 18 Pro with a 48-megapixel variable-aperture camera, while preorders began and delivery estimates for some models slipped past launch day. [1](https://www.apple.com/newsroom/2026/09/apple-debuts-iphone-18-pro-and-iphone-18-pro-max/) [6](https://www.macrumors.com/2026/09/14/iphone-18-pro-pre-orders-shipping-delivery-dates/)
> - 2026-09-22 to 2026-09-28: Early reliability scrutiny | Owners reported Face ID failures, freezing, and setup errors, though the reports were limited and no cause had been identified. [2](https://www.theapplepost.com/2026/09/22/72611/some-iphone-18-pro-users-report-face-id-problems-after-launch/)
> - 2026-10-06 to 2026-10-12: Demand narrative fractures | Reports of supplier order reductions attributed to softer demand were met by JPMorgan’s view that the adjustment was typical and China demand was stronger year over year. [4](https://finance.yahoo.com/technology/ai/articles/apple-cuts-iphone-18-pro-181700565.html) [7](https://www.gsmarena.com/jp_morgan_demand_for_iphone_18_pro_models_is_stronger_than_for_the_17_pro_models_last_year-news-75005.php)
> - 2026-10-20 to 2026-10-26: Supply and repair take focus | Coverage highlighted Apple’s expanded U.S. supplier partnerships and the availability of iPhone 18 Pro parts through its Self Service Repair store. [5](https://www.macrumors.com/2026/10/20/apple-touts-iphone-duo-18-pro-parts-made-in-usa/) [8](https://www.macrumors.com/2026/10/20/apple-iphone-18-pro-m6-mac-mini-repair-parts/)
>
> ## Contradictions
>
> - 2026-10-06 to 2026-10-12: Reports linked supplier order cuts to softer demand, while JPMorgan described the adjustment as typical and said China demand was stronger than for the prior-year Pro models. [4](https://finance.yahoo.com/technology/ai/articles/apple-cuts-iphone-18-pro-181700565.html) [7](https://www.gsmarena.com/jp_morgan_demand_for_iphone_18_pro_models_is_stronger_than_for_the_17_pro_models_last_year-news-75005.php)
> - 2026-10-13 to 2026-10-19: Jefferies said demand was weakening, while JPMorgan’s assessment described demand as modestly stronger than for the iPhone 17 Pro at the comparable point. [9](https://uk.finance.yahoo.com/news/apple-stock-dips-iphone-18-171334625.html) [10](https://finance.yahoo.com/markets/stocks/articles/jp-morgan-iphone-18-pro-154450108.html)
>
>
> # Coverage numbers, computed in the sandbox
>
> | window | results | distinct outlets |
> |---|---:|---:|
> | 2026-09-01 to 2026-09-07 | 13 | 7 |
> | 2026-09-08 to 2026-09-14 | 14 | 10 |
> | 2026-09-15 to 2026-09-21 | 13 | 7 |
> | 2026-09-22 to 2026-09-28 | 15 | 14 |
> | 2026-09-29 to 2026-10-05 | 17 | 14 |
> | 2026-10-06 to 2026-10-12 | 15 | 13 |
> | 2026-10-13 to 2026-10-19 | 19 | 19 |
> | 2026-10-20 to 2026-10-26 | 20 | 15 |
>
> Recurring outlets: 9to5mac.com (4), apple.com (3), cnet.com (3), cultofmac.com (3),
> finance.yahoo.com (3), gsmarena.com (3), macdailynews.com (3), macrumors.com (7)
>
> The Python the model ran is saved next to this file as coverage.py.
> ```

## Check the result against the sources

Three dated claims from the run, checked against the pages it cited:

* **September 9 launch.** The Sep 08 to Sep 14 window says Apple announced the iPhone 18 Pro on September 9 with a 48-megapixel variable-aperture camera, with preorders from September 12. [Apple's press release](https://www.apple.com/newsroom/2026/09/apple-debuts-iphone-18-pro-and-iphone-18-pro-max/), dated September 9, 2026, describes the "48MP Fusion Main camera with variable aperture" and says "Pre-orders begin Saturday, September 12." Correct.
* **October 9 order cut.** The Oct 06 to Oct 12 window says Apple asked suppliers to cut October component orders "by at least 15%, citing softer demand and higher memory-chip costs," and notes Apple did not confirm it. The cited [Indian Express report](https://indianexpress.com/article/technology/apple-cuts-iphone-18-pro-orders-due-to-soft-demand-report-10914405/), dated October 9, 2026, carries a Reuters story with "slashed by at least 15% ... Nikkei reported" and "Reuters could not immediately verify the report." Correct, hedge included.
* **October 20 supplier announcement.** The Oct 20 to Oct 26 window names Qnity Electronics as producing films for the variable-aperture camera and cites "nearly 100 components" from U.S. suppliers. The cited [MacRumors article](https://www.macrumors.com/2026/10/20/apple-touts-iphone-duo-18-pro-parts-made-in-usa/), dated October 20, 2026, says "Delaware-based Qnity Electronics will produce 'advanced films' for the variable aperture camera system on the iPhone 18 Pro models." Correct.

Two things to keep in mind when you read a result:

* A fast effort request makes one search and leads with one story. The Sep 22 to Sep 28 window led with the Face ID reports, not the iOS 27.0.1 fix that closed them on September 28.
* The sandbox table counts outlets among the results the search returned, capped at 20 per call. "19 distinct outlets" means the sample was spread across many sites, not that only 19 sites covered it.

## Compare effort levels on one window

`compare.py` sends one window's prompt and filters at three effort levels. Here is the Sep 22 to Sep 28 window at `effort="fast"`, `effort="medium"`, and `effort="high"`, run at the same moment:

```bash theme={null}
python compare.py "iPhone 18 Pro" --from 2026-09-22 --to 2026-09-28 --effort fast medium high
```

| | fast\@2026-10-16.0 | medium\@2026-10-22.1 | high\@2026-10-22.0 |
| - | - | - | - |
| total tokens | 8,842 | 20,019 | 21,040 |
| cost | \$0.0018 | \$0.0037 | \$0.0405 |
| tool result items | 1 | 2 | 4 |
| distinct dated sources | 13 | 30 | 18 |
| sources cited | 3 | 2 | 3 |
| headline | Early iPhone 18 Pro owners report Face ID problems | Apple released iOS 27.0.1 to fix iPhone 18 Pro bugs | iPhone 18 Pro reviews highlight performance, cooling and battery gains |

The "tool result items" row counts `search_results` and `fetch_url_results` items, the closest the response gets to a step count. The medium level ran two searches, saw 30 distinct dated pages, and led with the software fix the fast request missed. For a window where the story turns, that second search is what you pay for. The high level ran four tool calls on a window one search covers, at about 22 times the fast cost.

Five fast runs of this window on the same day led with the Face ID reports four times and the iOS 27.0.1 fix once. Both were in the week. If one window matters on its own, re-run it at the medium level for about twice the cost.

## Run the offline tests

`test_timeline.py` and `test_report.py` need no network or API key. Fake clients stand in for the API, including one that raises `RateLimitError` and one that returns a `sandbox_results` item. The tests cover the window arithmetic, the filter format, citation extraction, the headline split, the turning-point parser, the semaphore cap, the retry path, the Markdown and JSON writers, the sandbox request shape, the chart, and the summary table.

```bash theme={null}
pytest -q
```

```text theme={null}
17 passed
```

> **Read test\_timeline.py (full file, 189 lines)**
>
> ```python test_timeline.py theme={null}
> """Offline tests for timeline.py: windows, filters, parsing, retries. No network, no API key."""
>
> import asyncio
> from datetime import date
> from types import SimpleNamespace as NS
>
> import httpx
> import pytest
> from perplexity import RateLimitError
>
> import report
> import timeline as tl
>
>
> def window(i, start, end):
>     return tl.Window(i, date.fromisoformat(start), date.fromisoformat(end))
>
>
> def test_split_windows_short_last_window():
>     windows = tl.split_windows(date(2026, 9, 1), date(2026, 9, 17), 7)
>     assert [(w.start.day, w.end.day) for w in windows] == [(1, 7), (8, 14), (15, 17)]
>     assert [w.index for w in windows] == [1, 2, 3]
>
>
> def test_split_windows_step_larger_than_range():
>     windows = tl.split_windows(date(2026, 9, 1), date(2026, 9, 3), 30)
>     assert len(windows) == 1
>     assert (windows[0].start, windows[0].end) == (date(2026, 9, 1), date(2026, 9, 3))
>
>
> def test_split_windows_rejects_bad_input():
>     with pytest.raises(ValueError):
>         tl.split_windows(date(2026, 9, 3), date(2026, 9, 1), 7)
>     with pytest.raises(ValueError):
>         tl.split_windows(date(2026, 9, 1), date(2026, 9, 3), 0)
>
>
> def test_filters_use_mdy():
>     assert window(1, "2026-09-01", "2026-09-07").filters() == {
>         "search_after_date_filter": "09/01/2026",
>         "search_before_date_filter": "09/07/2026",
>     }
>
>
> def result(id, url, d, last="2026-10-26"):
>     return NS(id=id, url=url, title=f"title {id}", date=d, last_updated=last, snippet="")
>
>
> def fake_response(text, results, effort="fast@2026-10-16.0", cost=0.00169, tokens=6674):
>     usage = NS(total_tokens=tokens, cost=NS(total_cost=cost) if cost is not None else None)
>     return NS(
>         output=[
>             NS(type="search_results", results=results),
>             NS(type="message", content=[NS(type="output_text", text=text)]),
>         ],
>         usage=usage,
>         effort=effort,
>     )
>
>
> RESULTS = [
>     result(1, "https://a.example/1", "2026-09-05"),
>     result(2, "https://b.example/2", "2026-09-09"),  # outside the window
>     result(3, "https://c.example/3", None),  # undated
>     result(4, "https://a.example/1", "2026-09-06"),  # same URL as 1
>     result(5, "https://d.example/5", "2026-09-07"),
> ]
>
>
> def test_collect_maps_markers_to_result_ids_and_counts_dates():
>     w = window(1, "2026-09-01", "2026-09-07")
>     text, sources, citations, dated, undated = tl.collect(
>         fake_response("One thing happened[1][5]. Another[9].", RESULTS), w)
>     assert text.startswith("One thing")
>     assert len(sources) == 5
>     assert [c["id"] for c in citations] == [1, 5]
>     assert citations[0]["date"] == "2026-09-05"
>     assert dated == 2  # distinct URLs dated inside the window: a.example/1 and d.example/5
>     assert undated == 1
>
>
> def test_collect_without_window_skips_date_count_and_reads_web_markers():
>     text, sources, citations, dated, undated = tl.collect(fake_response("x[web:2] y[5]", RESULTS))
>     assert dated == 0 and [c["id"] for c in citations] == [2, 5]
>
>
> def test_turning_point_labels_lose_bold_markers():
>     class Stream:
>         def __init__(self):
>             self.events = [NS(type="response.output_text.delta", delta="x"),
>                            NS(type="response.completed", response=fake_response(
>                                "## Turning points\n- 2026-09-01 to 2026-09-07: **Slips** | d\n", []))]
>
>         def __aiter__(self):
>             return self
>
>         async def __anext__(self):
>             if not self.events:
>                 raise StopAsyncIteration
>             return self.events.pop(0)
>
>     class Client:
>         responses = NS(create=None)
>
>     async def create(**kwargs):
>         assert kwargs["stream"] is True and kwargs["effort"] == "medium"
>         assert kwargs["tools"][0]["filters"]["search_after_date_filter"] == "09/01/2026"
>         return Stream()
>
>     client = Client()
>     client.responses.create = create
>     out = []
>     results = [tl.WindowResult(window(1, "2026-09-01", "2026-09-07"), "Text[1].", "Headline", [],
>                                [tl.source_record(RESULTS[0])], 1, 0, 1.0, 100, 0.001, "fast@x")]
>     synth = asyncio.run(tl.run_synthesis(client, "t", results, NS(write=out.append,
>                                                                  flush=lambda: None)))
>     assert out == ["x"] and synth.turning_points == [(date(2026, 9, 1), date(2026, 9, 7), "Slips")]
>
>
> def test_turning_points_ignore_contradictions_section():
>     text = ("## Narrative\nx\n## Turning points\n- 2026-09-01 to 2026-09-07: Delay reported | Detail [1].\n"
>             "- 2026-09-15 to 2026-09-21: No pipe here\n- bad line\n"
>             "## Contradictions\n- 2026-09-22 to 2026-09-28: Not a turning point | d\n")
>     assert report.turning_points(text) == [
>         (date(2026, 9, 1), date(2026, 9, 7), "Delay reported"),
>         (date(2026, 9, 15), date(2026, 9, 21), "No pipe here")]
>
>
> class FakeClient:
>     """Counts concurrent calls and can fail with 429 a set number of times."""
>
>     def __init__(self, fail_first=0, delay=0.02):
>         self.fail_first, self.delay = fail_first, delay
>         self.active = self.peak = self.calls = 0
>         self.responses = NS(create=self.create)
>
>     async def create(self, **kwargs):
>         self.calls += 1
>         if self.calls <= self.fail_first:
>             resp = httpx.Response(429, headers={"retry-after": "0"},
>                                   request=httpx.Request("POST", "https://api.perplexity.ai"))
>             raise RateLimitError("slow down", response=resp, body=None)
>         self.active += 1
>         self.peak = max(self.peak, self.active)
>         await asyncio.sleep(self.delay)
>         self.active -= 1
>         return fake_response("Headline: Something happened\nSomething[1].", RESULTS)
>
>
> def test_semaphore_limits_concurrency():
>     client = FakeClient()
>     windows = tl.split_windows(date(2026, 9, 1), date(2026, 10, 26), 7)
>     sem = asyncio.Semaphore(3)
>
>     async def run():
>         return await asyncio.gather(
>             *[tl.run_window(client, sem, "topic", w, lambda _: None) for w in windows])
>
>     results = asyncio.run(run())
>     assert len(results) == 8 and client.peak <= 3
>     assert all(r.effort == "fast@2026-10-16.0" and r.cost == 0.00169 for r in results)
>     assert results[0].headline == "Something happened" and results[0].text == "Something[1]."
>
>
> def test_split_headline_handles_bold_markers_and_missing_line():
>     assert report.split_headline("**Headline:** Apple unveils the phone [1].\n\nBody[1].") == (
>         "Apple unveils the phone", "Body[1].")
>     assert report.split_headline("Headline: Nothing found\nNo dated pages.") == (
>         "Nothing found", "No dated pages.")
>     headline, body = report.split_headline("First sentence here[2]. Second one[3].")
>     assert headline == "First sentence here" and body.startswith("First sentence")
>
>
> def test_retry_after_429_then_success():
>     client, logged = FakeClient(fail_first=2), []
>     sem = asyncio.Semaphore(1)
>     r = asyncio.run(tl.run_window(client, sem, "topic", window(1, "2026-09-01", "2026-09-07"),
>                                   logged.append))
>     assert r.retries == 2 and client.calls == 3
>     assert len(logged) == 2 and "429" in logged[0]
>
>
> def test_retry_gives_up_after_limit():
>     client = FakeClient(fail_first=5)
>     sem = asyncio.Semaphore(1)
>     with pytest.raises(RateLimitError):
>         asyncio.run(tl.run_window(client, sem, "topic", window(1, "2026-09-01", "2026-09-07"),
>                                   lambda _: None))
>     assert client.calls == 3
> ```

> **Read test\_report.py (full file, 114 lines)**
>
> ```python test_report.py theme={null}
> """Offline tests for report.py and coverage.py: files, chart, table, sandbox step. No network."""
>
> import asyncio
> import json
> from datetime import date
> from types import SimpleNamespace as NS
>
> import coverage
> import report
> import timeline as tl
>
>
> def result(id, url, d, last="2026-10-26"):
>     return NS(id=id, url=url, title=f"title {id}", date=d, last_updated=last, snippet="")
>
>
> RESULT = result(1, "https://a.example/1", "2026-09-05")
>
>
> def sample_results():
>     windows = tl.split_windows(date(2026, 9, 1), date(2026, 9, 21), 7)
>     out = []
>     for i, w in enumerate(windows):
>         r = tl.WindowResult(w, f"Week {w.index} text[1].", f"Week {w.index} headline",
>                             [tl.source_record(RESULT)],
>                             [tl.source_record(RESULT)], dated_in_window=i + 2, latency=1.0 + i, tokens=6000, cost=0.0017,
>                             effort="fast@2026-10-16.0")
>         out.append(r)
>     return out
>
>
> def test_write_markdown_without_synthesis(tmp_path):
>     report.write_markdown(tmp_path / "timeline.md", "Topic", sample_results())
>     md = (tmp_path / "timeline.md").read_text()
>     assert "## Sep 15 to Sep 21 (4 dated sources)" in md and "How the story turned" not in md
>     assert "**Week 3 headline**" in md
>
>
> def test_week_by_week_and_windows_json(tmp_path):
>     results = sample_results()
>     block = report.week_by_week(results).splitlines()
>     assert block == ["Sep 01 to Sep 07  Week 1 headline", "Sep 08 to Sep 14  Week 2 headline",
>                      "Sep 15 to Sep 21  Week 3 headline"]
>     report.write_windows_json(tmp_path / "windows.json", results)
>     records = json.loads((tmp_path / "windows.json").read_text())
>     assert [r["start"] for r in records] == ["2026-09-01", "2026-09-08", "2026-09-15"]
>     assert records[0] == {
>         "index": 1, "start": "2026-09-01", "end": "2026-09-07", "headline": "Week 1 headline",
>         "summary": "Week 1 text[1].", "sources": [tl.source_record(RESULT)],
>         "results": [tl.source_record(RESULT)],
>         "dated_in_window": 2, "undated": 0, "latency_s": 1.0, "tokens": 6000, "cost": 0.0017,
>         "effort": "fast@2026-10-16.0"}
>
>
> def test_write_markdown_and_chart(tmp_path):
>     results = sample_results()
>     synth = tl.Synthesis(
>         text="## Narrative\nIt moved.\n## Turning points\n- 2026-09-08 to 2026-09-14: banks named\n",
>         citations=[{"id": 1, "url": "https://e.example", "title": "E", "date": None}],
>         turning_points=[(date(2026, 9, 8), date(2026, 9, 14), "banks named")],
>         latency=20.0, tokens=9000, cost=0.03, effort="medium@2026-10-16.0")
>     cov = coverage.Coverage(text="| window | results |", code="print(1)")
>     report.write_markdown(tmp_path / "timeline.md", "Topic", results, synth, cov)
>     md = (tmp_path / "timeline.md").read_text()
>     assert "## Sep 01 to Sep 07 (2 dated sources)" in md
>     assert "[title 1](https://a.example/1)" in md and "# How the story turned" in md
>     assert "undated [E](https://e.example)" in md
>     assert md.rstrip().endswith("# Coverage numbers, computed in the sandbox\n\n| window | results |\n\n"
>                                 "The Python the model ran is saved next to this file as coverage.py.")
>     report.draw_chart(tmp_path / "timeline.png", results, synth.turning_points, "Topic")
>     png = (tmp_path / "timeline.png").read_bytes()
>     assert png[:8] == b"\x89PNG\r\n\x1a\n" and len(png) > 5000
>
>
> def test_summary_table_and_slug():
>     cov = coverage.Coverage(latency=4.0, tokens=500, cost=0.0317)
>     table = report.summary_table(sample_results(), tl.Synthesis(latency=2, tokens=10, cost=None,
>                                                                 effort="medium@x"), cov, 12.0)
>     assert "2.0 s / 3.0 s" in table.splitlines()[2]
>     assert "n/a" in table and "wall time" in table and "4.0 s / 500 / $0.0317" in table
>     assert report.slugify("iPhone 18 Pro") == "iphone-18-pro"
>
>
> SANDBOX_CODE = "import json\nprint('macrumors.com 3')\n"
>
>
> class FakeSandboxClient:
>     """Returns the shape of a medium response that ran code in the sandbox."""
>
>     def __init__(self):
>         self.kwargs = None
>         self.responses = self
>
>     async def create(self, **kwargs):
>         self.kwargs = kwargs
>         run = NS(status="completed", exit_code=0, duration_ms=110, stdout="macrumors.com 3\n", stderr="")
>         items = [NS(type="sandbox_results", code=SANDBOX_CODE, language="python", results=[run]),
>                  NS(type="message", content=[NS(type="output_text", text="| window | results |")])]
>         usage = NS(total_tokens=15765, cost=NS(total_cost=0.0317),
>                    tool_calls_details={"sandbox": NS(cost_usd=0.03)})
>         return NS(output=items, usage=usage, effort="medium@2026-10-22.1")
>
>
> def test_run_coverage_reads_sandbox_items_and_sends_only_the_sandbox_tool():
>     client = FakeSandboxClient()
>     cov = asyncio.run(coverage.run_coverage(client, "Topic", sample_results()))
>     assert client.kwargs["effort"] == "medium" and client.kwargs["tools"] == [{"type": "sandbox"}]
>     assert "do not search the web" in client.kwargs["input"]
>     assert '{"window":"2026-09-01 to 2026-09-07","hosts":["a.example"]}' in client.kwargs["input"]
>     assert coverage.host("https://www.macrumors.com/2026/10/09/x/") == "macrumors.com"
>     assert cov.code == SANDBOX_CODE and cov.exit_code == 0 and cov.duration_ms == 110
>     assert cov.text == "| window | results |" and cov.tokens == 15765 and cov.sandbox_cost == 0.03
>     summary = coverage.coverage_summary(cov)
>     assert "2 lines of Python" in summary and "$0.0317 incl. $0.03 sandbox session" in summary
> ```

## Keep these limits in mind

* **The bounds are inclusive, as observed.** Results dated on both filter days came back in every probe. Windows here do not overlap because each starts the day after the last one ends. The reference page does not state inclusivity, so check a boundary date in your own run if you depend on it.
* **Undated pages do not come back under a publication-date filter.** No result arrived with `date: null` when the filters were on; the unfiltered probe returned one. Pages with no publication date in the index will not show up in your windows.
* **`last_updated` filters answer a different question.** Every page in the probes had a `last_updated` weeks after its `date`, because pages get re-indexed as they change. Use the `last_updated` pair for content that is current, such as documentation or pricing. Use the publication pair for what was said when.
* **The source count is a capped sample.** Each fast effort request asks for 20 results, so the bars top out at 20 and a window with 200 articles looks like one with 20. Read the bars as a density hint and the text for the rest.
* **Each window is blind to the others.** The request for September 22 knows nothing about October 13. The synthesis is the only place the story gets connected.
* **Cost scales with windows.** Eight windows cost $0.0141 at the fast effort level and sent 5,753 tokens into the synthesis. A 52-window year is about $0.09 of fast requests and a synthesis input of roughly 50,000 tokens; synthesize quarters first, then the quarters. The sandbox session costs \$0.03 whether you have 8 windows or 52.
* **The sandbox and date filters do not mix in one request.** Keep the filtered searches in their own requests and hand the sandbox step their data, as this recipe does.
* **The model only computes what it can read.** Data in the prompt reaches the sandbox only if the model writes it to a file. Tell it to save the input before parsing, keep the payload small (the hostnames prompt here is about 3 KB), and read the returned code once per effort version.
* **Effort levels are versioned.** This page used `fast@2026-10-16.0`, `medium@2026-10-22.1`, and `high@2026-10-22.0`. A new version can change the model, the default `max_results`, or the citation marker format, which is why `collect` accepts both `[3]` and `[web:3]` and the script prints the version strings.

## Read next

* Set `--step` to 1 for a day-by-day view of one busy week, or 30 for a year in twelve windows.
* Add `search_domain_filter` to the `filters` dictionary to restrict each window to a few publications and compare how two outlets covered the same weeks.
* Swap in `last_updated_after_filter` and `last_updated_before_filter` to ask which pages changed that week rather than what was published.
* Hand the sandbox step `windows.json` and ask for the day-by-day spread of publication dates inside each window, or the share of each window's results from the recurring outlets.
* Replace the `print` lines in `main` with writes to a database table keyed by topic and window, and run the script on a schedule so the timeline extends itself each week.
* Give the synthesis a [JSON schema](https://docs.perplexity.ai/docs/agent-api/building-agents/shape-output.md) when you want turning points as data rather than Markdown lines.

> **[Filter web search by date](https://docs.perplexity.ai/docs/agent-api/tools/web-search.md)**
>
> Every filter the search tool accepts, including domain, recency, and date ranges.

> **[Choose an effort level](https://docs.perplexity.ai/docs/agent-api/effort.md)**
>
> What each level bundles, how versions work, and how to pin one.

> **[Run code in the sandbox](https://docs.perplexity.ai/docs/agent-api/tools/sandbox.md)**
>
> How the model runs code in a container and what the response carries back.

> **[Give the agent a sandbox](https://docs.perplexity.ai/docs/agent-api/building-agents/run-code.md)**
>
> When to give the agent a sandbox so it computes exact answers instead of estimating them.