baseline flags (migration 0002) so first-snapshot entities never count as new; SSE no-transform; loopback SSR = internal rate tier
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 changed files +56 −12
added
migrations/versions/0002_baseline_flags.py
+39 −0
@@ -0,0 +1,39 @@ | ||
| 1 | +"""Baseline flags on extracted entities + trigram search indexes. | |
| 2 | + | |
| 3 | +Entities captured by a sensor's very first snapshot are not "new" — they existed before we started observing (coverage | |
| 4 | +normalisation, spec §145). `baseline = true` lets metrics, daily aggregates and job summaries count only entities that appeared | |
| 5 | +*after* observation started, without touching first_seen_at (which stays the honest observation timestamp). | |
| 6 | + | |
| 7 | +Revision ID: 0002 | |
| 8 | +Revises: 0001 | |
| 9 | +""" | |
| 10 | +from __future__ import annotations | |
| 11 | + | |
| 12 | +from alembic import op | |
| 13 | + | |
| 14 | +revision = "0002" | |
| 15 | +down_revision = "0001" | |
| 16 | +branch_labels = None | |
| 17 | +depends_on = None | |
| 18 | + | |
| 19 | +SQL = """ | |
| 20 | +alter table jobs add column if not exists baseline boolean not null default false; | |
| 21 | +alter table people add column if not exists baseline boolean not null default false; | |
| 22 | +alter table products add column if not exists baseline boolean not null default false; | |
| 23 | +alter table locations add column if not exists baseline boolean not null default false; | |
| 24 | +alter table news_items add column if not exists baseline boolean not null default false; | |
| 25 | +alter table pricing_plans add column if not exists baseline boolean not null default false; | |
| 26 | +create index if not exists jobs_new_idx on jobs (company_id, first_seen_at desc) where not baseline; | |
| 27 | +create index if not exists news_items_new_idx on news_items (company_id, first_seen_at desc) where not baseline; | |
| 28 | +create index if not exists people_name_trgm_idx on people using gin (name gin_trgm_ops); | |
| 29 | +create index if not exists products_name_trgm_idx on products using gin (name gin_trgm_ops); | |
| 30 | +""" | |
| 31 | + | |
| 32 | + | |
| 33 | +def upgrade() -> None: | |
| 34 | + for stmt in [s.strip() for s in SQL.split(";") if s.strip()]: | |
| 35 | + op.execute(stmt) | |
| 36 | + | |
| 37 | + | |
| 38 | +def downgrade() -> None: | |
| 39 | + raise RuntimeError("forward-only migrations") | |
modified
src/companyatlas/api/routers/companies.py
+2 −2
@@ -69,7 +69,7 @@ async def compare(response: Response, companies: str = Query(..., description="c | ||
| 69 | 69 | "company_id = any(cast(:ids as text[])) group by 1, 2", ids=ids, d=q.days_ago(30)) |
| 70 | 70 | job_rows = await fetch_all(conn, "select company_id, count(*) filter (where status = 'open') as open, " |
| 71 | 71 | "count(*) filter (where status = 'open' and is_ai) as ai_open, " |
| 72 | − "count(*) filter (where first_seen_at >= :d) as new_30d from jobs where company_id = any(cast(:ids as text[])) " | |
| 72 | + "count(*) filter (where first_seen_at >= :d and not baseline) as new_30d from jobs where company_id = any(cast(:ids as text[])) " | |
| 73 | 73 | "group by company_id", ids=ids, d=q.days_ago(30)) |
| 74 | 74 | loc_rows = await fetch_all(conn, "select company_id, count(*) as n from locations where status = 'listed' and company_id = any(cast(:ids as text[])) " |
| 75 | 75 | "group by company_id", ids=ids) |
@@ -231,7 +231,7 @@ async def company_jobs(key: str, response: Response, p: PageDep, status: str = Q | ||
| 231 | 231 | rows = await fetch_all(conn, f"select j.* from jobs j where {wsql} order by {order} limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset) |
| 232 | 232 | total = await q.bounded_count(conn, f"from jobs j where {wsql}", params) |
| 233 | 233 | d7 = q.days_ago(7) |
| 234 | − s = await fetch_one(conn, "select count(*) filter (where status = 'open') as open, count(*) filter (where first_seen_at >= :d7 and status = 'open') as new_7d, " | |
| 234 | + s = await fetch_one(conn, "select count(*) filter (where status = 'open') as open, count(*) filter (where first_seen_at >= :d7 and status = 'open' and not baseline) as new_7d, " | |
| 235 | 235 | "count(*) filter (where removed_at >= :d7) as removed_7d, count(*) filter (where status = 'open' and is_ai) as ai_open, " |
| 236 | 236 | "count(*) filter (where status = 'open' and remote is true) as remote_open, " |
| 237 | 237 | "count(*) filter (where status = 'open' and remote is not null) as remote_known from jobs where company_id = :id", |
modified
src/companyatlas/api/sse.py
+1 −1
@@ -25,7 +25,7 @@ POLL_S = 5.0 | ||
| 25 | 25 | HEARTBEAT_S = 20.0 |
| 26 | 26 | BATCH = 50 |
| 27 | 27 | MAX_STREAM_S = 3600 |
| 28 | −SSE_HEADERS = {"cache-control": "no-store", "x-accel-buffering": "no", "connection": "keep-alive"} | |
| 28 | +SSE_HEADERS = {"cache-control": "no-store, no-transform", "x-accel-buffering": "no", "connection": "keep-alive"} | |
| 29 | 29 | |
| 30 | 30 | |
| 31 | 31 | def _dump(payload: Any) -> str: |
modified
src/companyatlas/services/metrics.py
+9 −9
@@ -195,25 +195,25 @@ async def load_context(conn, company_id: str, now: datetime) -> CompanyContext | | ||
| 195 | 195 | count(*) filter (where first_seen_at <= :t7 and (removed_at is null or removed_at > :t7)) as open_7, |
| 196 | 196 | count(*) filter (where first_seen_at <= :t30 and (removed_at is null or removed_at > :t30)) as open_30, |
| 197 | 197 | count(*) filter (where first_seen_at <= :t90 and (removed_at is null or removed_at > :t90)) as open_90, |
| 198 | − count(*) filter (where first_seen_at > :t30) as new_30d, | |
| 198 | + count(*) filter (where first_seen_at > :t30 and not baseline) as new_30d, | |
| 199 | 199 | count(*) filter (where removed_at > :t30) as removed_30d, |
| 200 | − count(*) filter (where first_seen_at > :t30 and is_ai) as ai_new_30d, | |
| 200 | + count(*) filter (where first_seen_at > :t30 and is_ai and not baseline) as ai_new_30d, | |
| 201 | 201 | count(*) as total |
| 202 | 202 | from jobs where company_id = :c""", c=company_id, t7=now - timedelta(days=7), t30=d30, t90=d90) |
| 203 | 203 | by_country = await fetch_all(conn, "select country, count(*) as n from jobs where company_id = :c and status = 'open' and country is not null group by country order by n desc limit 20", c=company_id) |
| 204 | 204 | by_department = await fetch_all(conn, "select department, count(*) as n from jobs where company_id = :c and status = 'open' and department is not null group by department order by n desc limit 20", c=company_id) |
| 205 | 205 | new_job_countries = await fetch_all(conn, """ |
| 206 | − select country from jobs where company_id = :c and country is not null group by country having min(first_seen_at) > :since""", c=company_id, since=d90) | |
| 206 | + select country from jobs where company_id = :c and country is not null group by country having min(first_seen_at) > :since and bool_and(not baseline)""", c=company_id, since=d90) | |
| 207 | 207 | products = await fetch_one(conn, """select count(*) filter (where status = 'listed') as listed, count(*) as total, |
| 208 | 208 | array_agg(name) filter (where status = 'listed') as names from products where company_id = :c""", c=company_id) |
| 209 | − news = await fetch_one(conn, """select count(*) filter (where coalesce(published_at, first_seen_at) >= :d30) as n_30d, | |
| 210 | − count(*) as total, array_agg(title) filter (where coalesce(published_at, first_seen_at) >= :d90) as titles_90d | |
| 209 | + news = await fetch_one(conn, """select count(*) filter (where coalesce(published_at, first_seen_at) >= :d30 and not (baseline and published_at is null)) as n_30d, | |
| 210 | + count(*) as total, array_agg(title) filter (where coalesce(published_at, first_seen_at) >= :d90 and not (baseline and published_at is null)) as titles_90d | |
| 211 | 211 | from news_items where company_id = :c""", c=company_id, d30=d30, d90=d90) |
| 212 | 212 | locations = await fetch_one(conn, """ |
| 213 | − select count(*) filter (where status = 'listed') as listed, count(*) filter (where first_seen_at > :since) as new_90d, count(*) as total, | |
| 213 | + select count(*) filter (where status = 'listed') as listed, count(*) filter (where first_seen_at > :since and not baseline) as new_90d, count(*) as total, | |
| 214 | 214 | count(distinct country) filter (where status = 'listed' and country is not null) as countries |
| 215 | 215 | from locations where company_id = :c""", c=company_id, since=d90) |
| 216 | − new_loc_countries = await fetch_all(conn, "select country from locations where company_id = :c and country is not null group by country having min(first_seen_at) > :since", | |
| 216 | + new_loc_countries = await fetch_all(conn, "select country from locations where company_id = :c and country is not null group by country having min(first_seen_at) > :since and bool_and(not baseline)", | |
| 217 | 217 | c=company_id, since=d90) |
| 218 | 218 | titles = await fetch_all(conn, """select distinct on (s.sensor_id) s.title from snapshots s join sensors se on se.id = s.sensor_id |
| 219 | 219 | where se.company_id = :c order by s.sensor_id, s.fetched_at desc""", c=company_id) |
@@ -428,7 +428,7 @@ async def compute_daily(day: date) -> dict[str, Any]: | ||
| 428 | 428 | b = bucket(r["company_id"]) |
| 429 | 429 | b["events"] += int(r["n"]) |
| 430 | 430 | b["events_by_type"][r["event_type"]] = int(r["n"]) |
| 431 | − for r in await fetch_all(conn, """select company_id, count(*) filter (where first_seen_at >= :d0 and first_seen_at < :d1) as new_n, | |
| 431 | + for r in await fetch_all(conn, """select company_id, count(*) filter (where first_seen_at >= :d0 and first_seen_at < :d1 and not baseline) as new_n, | |
| 432 | 432 | count(*) filter (where removed_at >= :d0 and removed_at < :d1) as rem_n, |
| 433 | 433 | count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1)) as open_n, |
| 434 | 434 | count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1) and is_ai) as ai_n |
@@ -437,7 +437,7 @@ async def compute_daily(day: date) -> dict[str, Any]: | ||
| 437 | 437 | continue |
| 438 | 438 | b = bucket(r["company_id"]) |
| 439 | 439 | b["jobs_new"], b["jobs_removed"], b["jobs_open"], b["jobs_ai_open"] = int(r["new_n"]), int(r["rem_n"]), int(r["open_n"]), int(r["ai_n"]) |
| 440 | − for r in await fetch_all(conn, "select company_id, count(*) as n from news_items where first_seen_at >= :d0 and first_seen_at < :d1 group by 1", d0=d0, d1=d1): | |
| 440 | + for r in await fetch_all(conn, "select company_id, count(*) as n from news_items where first_seen_at >= :d0 and first_seen_at < :d1 and not baseline group by 1", d0=d0, d1=d1): | |
| 441 | 441 | bucket(r["company_id"])["news_items"] = int(r["n"]) |
| 442 | 442 | if per: |
| 443 | 443 | ids = list(per) |
modified
src/companyatlas/services/pipeline.py
+5 −0
@@ -490,6 +490,11 @@ async def reconcile(conn: Any, *, company_id: str, sensor: dict[str, Any], ex: E | ||
| 490 | 490 | if (previous_meta.get("description") or None) != (ex.meta.get("description") or None): |
| 491 | 491 | meta["description_changed"] = True |
| 492 | 492 | delta["meta"] = meta |
| 493 | + if int(sensor.get("snapshot_count") or 0) == 0: | |
| 494 | + # First snapshot of this sensor: everything it lists pre-dates our observation. Flag it so "new" counts stay honest (spec §145). | |
| 495 | + for table in ("jobs", "people", "products", "locations", "news_items", "pricing_plans"): | |
| 496 | + await execute(conn, f"update {table} set baseline = true where sensor_id = :sid and first_seen_at = :now", sid=sid, now=now) | |
| 497 | + delta["baseline"] = True | |
| 493 | 498 | return delta |
| 494 | 499 | |
| 495 | 500 | |
| 496 | 501 | |