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

"""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()))

"""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("-")

"""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}")

"""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()))

"""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

"""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