Sync ingestion baseline and noise suppression

src/db.ts

@@ -26,6 +26,7 @@ export async function migrate() {
       enabled BOOLEAN NOT NULL DEFAULT TRUE,
       last_synced_at TIMESTAMPTZ,
       last_error TEXT,
+      baseline_complete BOOLEAN NOT NULL DEFAULT FALSE,
       created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
     );
 
@@ -50,6 +51,7 @@ export async function migrate() {
       published_at TIMESTAMPTZ,
       detected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
       content_hash TEXT NOT NULL,
+      change_kind TEXT NOT NULL DEFAULT 'new',
       topics JSONB NOT NULL DEFAULT '[]'::jsonb,
       raw_json JSONB NOT NULL,
       UNIQUE(source_id, external_id, content_hash)
@@ -107,6 +109,12 @@ export async function migrate() {
     CREATE INDEX IF NOT EXISTS idx_agent_messages_unread
       ON agent_messages(acknowledged_at, created_at DESC);
 
+    CREATE TABLE IF NOT EXISTS app_meta (
+      key TEXT PRIMARY KEY,
+      value JSONB NOT NULL DEFAULT '{}'::jsonb,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+
     CREATE TABLE IF NOT EXISTS audit_events (
       id TEXT PRIMARY KEY,
       event_type TEXT NOT NULL,
@@ -116,6 +124,26 @@ export async function migrate() {
       created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
     );
   `);
+
+  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS baseline_complete BOOLEAN NOT NULL DEFAULT FALSE");
+  await pool.query("ALTER TABLE signals ADD COLUMN IF NOT EXISTS change_kind TEXT NOT NULL DEFAULT 'new'");
+
+  const marker = await pool.query(
+    `INSERT INTO app_meta(key,value)
+     VALUES('noise_baseline_v1','{"applied":true}'::jsonb)
+     ON CONFLICT(key) DO NOTHING
+     RETURNING key`
+  );
+
+  if (marker.rowCount) {
+    await pool.query(
+      `UPDATE opportunities
+       SET status='closed',
+           reasons = reasons || '["Archived during initial baseline cleanup"]'::jsonb,
+           updated_at=NOW()
+       WHERE status='new'`
+    );
+  }
 }
 
 export async function seedSources(sources: SourceDefinition[]) {
@@ -127,6 +155,13 @@ export async function seedSources(sources: SourceDefinition[]) {
       [source.id, source.name, source.type, source.url]
     );
   }
+
+  await pool.query(
+    `UPDATE sources s
+     SET baseline_complete=TRUE
+     WHERE baseline_complete=FALSE
+       AND EXISTS (SELECT 1 FROM signals sig WHERE sig.source_id=s.id)`
+  );
 }
 
 export function contentHash(item: NormalizedItem) {
@@ -141,35 +176,88 @@ export function contentHash(item: NormalizedItem) {
     .digest("hex");
 }
 
-export async function storeItem(item: NormalizedItem, scored: ScoredItem): Promise<{ created: boolean; signalId?: string; opportunityId?: string }> {
+export async function storeItem(
+  item: NormalizedItem,
+  scored: ScoredItem,
+  options: { baseline?: boolean } = {}
+): Promise<{
+  created: boolean;
+  signalId?: string;
+  opportunityId?: string;
+  changeKind?: "baseline" | "new" | "updated";
+}> {
   const hash = contentHash(item);
-  const existing = await pool.query(
+  const previous = await pool.query(
-    "SELECT id FROM signals WHERE source_id=$1 AND external_id=$2 AND content_hash=$3 LIMIT 1",
+    `SELECT id,content_hash
-    [item.sourceId, item.externalId, hash]
+     FROM signals
+     WHERE source_id=$1 AND external_id=$2
+     ORDER BY detected_at DESC
+     LIMIT 1`,
+    [item.sourceId, item.externalId]
   );
-  if (existing.rowCount) return { created: false };
+
+  if (previous.rowCount && previous.rows[0].content_hash === hash) {
+    return { created: false };
+  }
+
+  const changeKind: "baseline" | "new" | "updated" =
+    options.baseline ? "baseline" : previous.rowCount ? "updated" : "new";
 
   const signalId = randomUUID();
   const opportunityId = randomUUID();
   const evidenceId = randomUUID();
 
+  const minScore = Number(process.env.OPPORTUNITY_MIN_SCORE ?? 2);
+  const maxAgeDays = Math.max(Number(process.env.MAX_OPPORTUNITY_AGE_DAYS ?? 14), 1);
+
+  let recentEnough = true;
+  if (item.publishedAt) {
+    const published = new Date(item.publishedAt).getTime();
+    if (!Number.isNaN(published)) {
+      recentEnough = Date.now() - published <= maxAgeDays * 86_400_000;
+    }
+  }
+
+  const shouldCreateOpportunity =
+    !options.baseline &&
+    recentEnough &&
+    scored.score >= minScore;
+
   await pool.query("BEGIN");
   try {
     await pool.query(
-      `INSERT INTO signals(id,source_id,external_id,title,summary,url,published_at,content_hash,topics,raw_json)
+      `INSERT INTO signals(
-       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10::jsonb)`,
+         id,source_id,external_id,title,summary,url,published_at,content_hash,change_kind,topics,raw_json
+       )
+       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11::jsonb)`,
       [
-        signalId, item.sourceId, item.externalId, item.title, item.summary, item.url,
+        signalId,
-        item.publishedAt, hash, JSON.stringify(item.topics), JSON.stringify(item.raw)
+        item.sourceId,
+        item.externalId,
+        item.title,
+        item.summary,
+        item.url,
+        item.publishedAt,
+        hash,
+        changeKind,
+        JSON.stringify(item.topics),
+        JSON.stringify(item.raw)
       ]
     );
+
     await pool.query(
       `INSERT INTO evidence(id,signal_id,source_url,content_hash,excerpt)
        VALUES($1,$2,$3,$4,$5)`,
-      [evidenceId, signalId, item.url, hash, `${item.title}\n${item.summary}`.slice(0, 2000)]
+      [
+        evidenceId,
+        signalId,
+        item.url,
+        hash,
+        `${item.title}\n${item.summary}`.slice(0, 2000)
+      ]
     );
 
-    if (scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1)) {
+    if (shouldCreateOpportunity) {
       await pool.query(
         `INSERT INTO opportunities(id,signal_id,status,score,reasons)
          VALUES($1,$2,'new',$3,$4::jsonb)`,
@@ -180,10 +268,28 @@ export async function storeItem(item: NormalizedItem, scored: ScoredItem): Promi
     await pool.query(
       `INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
        VALUES($1,'signal.ingested','signal',$2,$3::jsonb)`,
-      [randomUUID(), signalId, JSON.stringify({ sourceId: item.sourceId, externalId: item.externalId, score: scored.score })]
+      [
+        randomUUID(),
+        signalId,
+        JSON.stringify({
+          sourceId: item.sourceId,
+          externalId: item.externalId,
+          score: scored.score,
+          changeKind,
+          opportunityCreated: shouldCreateOpportunity,
+          recentEnough
+        })
+      ]
     );
+
     await pool.query("COMMIT");
-    return { created: true, signalId, opportunityId: scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1) ? opportunityId : undefined };
+
+    return {
+      created: true,
+      signalId,
+      opportunityId: shouldCreateOpportunity ? opportunityId : undefined,
+      changeKind
+    };
   } catch (error) {
     await pool.query("ROLLBACK");
     throw error;

src/index.ts

@@ -61,7 +61,7 @@ app.get("/ready", async (_req, res) => {
               COUNT(sig.id)::int AS signal_count
        FROM sources s
        LEFT JOIN signals sig ON sig.source_id=s.id
-       GROUP BY s.id,s.name,s.last_synced_at,s.last_error
+       GROUP BY s.id,s.name,s.last_synced_at,s.last_error,s.baseline_complete
        ORDER BY s.id`
     );
     const missing = result.rows
@@ -114,7 +114,7 @@ app.post("/messages/:id/ack", requireWebAuth, async (req, res) => {
 
 app.get("/", requireWebAuth, async (req, res) => {
   const sourceRows = await pool.query(
-    `SELECT s.id,s.name,s.last_synced_at,s.last_error,
+    `SELECT s.id,s.name,s.last_synced_at,s.last_error,s.baseline_complete,
             COUNT(sig.id)::int AS signal_count
      FROM sources s
      LEFT JOIN signals sig ON sig.source_id=s.id
@@ -124,7 +124,7 @@ app.get("/", requireWebAuth, async (req, res) => {
   const counts = await pool.query(`
     SELECT
       (SELECT COUNT(*) FROM signals) AS signals,
-      (SELECT COUNT(*) FROM opportunities) AS opportunities,
+      (SELECT COUNT(*) FROM opportunities WHERE status <> 'closed') AS opportunities,
       (SELECT COUNT(*) FROM opportunities WHERE status='new') AS new_opportunities,
       (SELECT COUNT(*) FROM agent_messages WHERE acknowledged_at IS NULL) AS unread_messages
   `);
@@ -175,7 +175,7 @@ ${messages.length
     </div>`).join("")
   : '<div class="card"><span class="meta">No unread agent messages.</span></div>'}
 <h2>Sources</h2>
-${sourceRows.rows.map((s:any)=>`<div class="card"><strong>${s.name}</strong><br><code>${s.id}</code><br>${s.signal_count} stored signals<br><span class="${s.last_error?'bad':'ok'}">${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}</span></div>`).join("")}
+${sourceRows.rows.map((s:any)=>`<div class="card"><strong>${s.name}</strong><br><code>${s.id}</code> · ${s.baseline_complete ? 'baseline ready' : 'awaiting baseline'}<br>${s.signal_count} stored signals<br><span class="${s.last_error?'bad':'ok'}">${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}</span></div>`).join("")}
 <p>MCP endpoint: <code>/api/mcp</code> — OAuth protected separately from this browser session.</p>
 </body></html>`);
 });

src/sync.ts

@@ -15,21 +15,41 @@ export async function syncSource(sourceId: SourceId) {
     [runId, sourceId]
   );
   try {
+    const sourceState = await pool.query(
+      "SELECT baseline_complete FROM sources WHERE id=$1",
+      [sourceId]
+    );
+    const baseline = !Boolean(sourceState.rows[0]?.baseline_complete);
+
     const items = await fetchSource(sourceId);
     let created = 0;
+    let opportunitiesCreated = 0;
+
     for (const item of items) {
-      const result = await storeItem(item, scoreItem(item));
+      const result = await storeItem(item, scoreItem(item), { baseline });
       if (result.created) created++;
+      if (result.opportunityId) opportunitiesCreated++;
     }
     await pool.query(
       "UPDATE source_runs SET status='ok',finished_at=NOW(),item_count=$2,new_signal_count=$3 WHERE id=$1",
       [runId, items.length, created]
     );
     await pool.query(
-      "UPDATE sources SET last_synced_at=NOW(),last_error=NULL WHERE id=$1",
+      `UPDATE sources
-      [sourceId]
+       SET last_synced_at=NOW(),
+           last_error=NULL,
+           baseline_complete=CASE WHEN $2 THEN TRUE ELSE baseline_complete END
+       WHERE id=$1`,
+      [sourceId, baseline && items.length > 0]
     );
-    return { sourceId, items: items.length, newSignals: created };
+
+    return {
+      sourceId,
+      mode: baseline ? "baseline" : "live",
+      items: items.length,
+      newSignals: created,
+      opportunitiesCreated
+    };
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
     await pool.query(