diff --git a/dashboard/analysis_handoff.js b/dashboard/analysis_handoff.js new file mode 100644 index 00000000..b18dd03b --- /dev/null +++ b/dashboard/analysis_handoff.js @@ -0,0 +1,105 @@ +(() => { + 'use strict'; + const $ = s => document.querySelector(s); + const esc = v => String(v ?? '').replace(/[&<>"']/g, c => ({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c])); + let period = 'week', start = '', end = '', preview = null, generation = 0; + const dateInput = d => `${d.getFullYear()}-${String(d.getMonth()+1).padStart(2,'0')}-${String(d.getDate()).padStart(2,'0')}`; + function selection() { + if (period === 'current' || period === 'all') return {scope: period}; + const now = new Date(), a = new Date(now.getFullYear(), now.getMonth(), now.getDate()); + const b = new Date(a); b.setDate(b.getDate()+1); + if (period === 'week') a.setDate(a.getDate()-6); + if (period === 'custom') { + if (!start || !end) throw new Error('Choose both dates.'); + const aa = new Date(start+'T00:00:00'), bb = new Date(end+'T00:00:00'); + bb.setDate(bb.getDate()+1); + if (!Number.isFinite(+aa) || !Number.isFinite(+bb) || aa >= bb) throw new Error('End date must be on or after start.'); + return {scope:'range',since:aa.toISOString(),until:bb.toISOString()}; + } + return {scope:'range',since:a.toISOString(),until:b.toISOString()}; + } + window.owgEvidenceSelection = selection; + async function api(path, options) { + await window.__owgAuthReady; + const r = await fetch(path,{cache:'no-store',...options}); + const d = await r.json().catch(()=>({})); + if (!r.ok) throw new Error(d.detail || 'Could not prepare evidence.'); + return d; + } + async function download(fmt, detailed, redacted=true, chosen=selection()) { + const d = await api('/v1/export-ticket',{method:'POST',headers:{'Content-Type':'application/json'}, + body:JSON.stringify({format:fmt,...chosen,include_raw:!!detailed,redact_names:!!redacted})}); + const a = document.createElement('a');a.href=d.url;a.download='';a.hidden=true; + document.body.appendChild(a);a.click();a.remove(); + } + function syncControls() { + document.querySelectorAll('.owg-period').forEach(node => { + node.querySelector('select').value=period; + node.querySelector('[data-custom]').hidden=period!=='custom'; + node.querySelector('[data-start]').value=start;node.querySelector('[data-end]').value=end; + }); + } + async function refresh() { + const seq=++generation; preview=null; + document.querySelectorAll('[data-period-summary]').forEach(x=>x.textContent='Checking selected evidence…'); + try { + const d=await api('/v1/analysis-preview?'+new URLSearchParams(selection())); + if(seq!==generation)return;preview=d; + const text=d.canonical_rows + ? `${d.canonical_rows} recorded events · observed ${new Date(d.first_observed_at).toLocaleString()} – ${new Date(d.last_observed_at).toLocaleString()}` + : d.retained_rows ? 'No activity in this selection. Saved work is available; choose another period.' : 'No saved evidence yet. Work normally after enabling recording.'; + document.querySelectorAll('[data-period-summary]').forEach(x=>x.textContent=text); + } catch(e) { + if(seq!==generation)return; + document.querySelectorAll('[data-period-summary]').forEach(x=>x.textContent=e.message); + } + } + async function prepare() { + await refresh(); + if(!preview){window.toast?.('Could not prepare selected evidence. Check the period and dashboard connection.');return;} + const d=preview, request=d.request_text; + const chosen={scope:d.window.scope,...(d.window.scope==='range'?{since:d.window.since,until:d.window.until}:{})}; + const access=d.local_ai_history_access || {}; + const localRead=!d.local_ai_enabled ? 'Local AI access is off.' : access.mode==='off' ? 'Local AI can read the current run only; saved-history access is off.' : `Local saved-history permission: ${access.mode}${access.expires_at?' until '+new Date(access.expires_at).toLocaleString():''}.`; + window.openModal?.('Prepare for AI','Discover opportunities from recorded work',` +

No workflow description or repeated pattern is required. Your AI will interpret the evidence; this panel does not run a model.

+
What the AI will receive +

${esc(d.canonical_rows)} recorded events · ${esc(d.event_types.focus_span||0)} focus spans · ${esc(d.event_types.capture_gap||0)} unobserved-gap records.

+

Selected: ${esc(d.window.since||'all available history')} to ${esc(d.window.until||'the latest retained evidence')} (end exclusive, UTC). Observed: ${esc(d.first_observed_at||'none')} to ${esc(d.last_observed_at||'none')}.

+

Sources: ${esc(Object.entries(d.sources).map(([k,v])=>`${k}: ${v}`).join(', ')||'none')}. Visible work text is not included in this activity package. Typed values and clipboard contents are not included.

+

${esc(localRead)} Hosted AI availability must be checked separately: it only sees previously shared evidence. Preparing a file does not change those permissions.

+
+
Preview redacted context examples

Filtering is best effort; inspect the downloaded package before sharing it.

+ +

Downloaded evidence is redacted. Gaps and foreground time are not measured lost work or savings.

+
+
Read the request

+      
Nothing has been sent to an AI. Downloads work with any capable AI, without an OWG cloud account.
`);
+    $('#analysisRequest').textContent=request;
+    $('#analysisCopy').onclick=async()=>{
+      try{await navigator.clipboard.writeText(request);$('#analysisActionStatus').textContent='Request copied. Paste it into your AI; no model was invoked.';}
+      catch(_){$('#analysisActionStatus').textContent='Clipboard unavailable. Open “Read the request” and copy the text.';}
+    };
+    $('#analysisDownload').onclick=async()=>{
+      try{await download('csvzip',$('#analysisDetailed').checked,true,chosen);$('#analysisActionStatus').textContent='Download started. Inspect the file, then attach it to your AI. No data was uploaded.';}
+      catch(e){$('#analysisActionStatus').textContent=e.message;}
+    };
+  }
+  function install() {
+    const style=document.createElement('style');style.textContent='.owg-period{display:flex;align-items:center;flex-wrap:wrap;gap:10px;margin-bottom:14px;padding:12px 14px;border:1px solid var(--line);border-radius:12px;background:#fff}.owg-period select,.owg-period input{font:inherit;padding:7px;border:1px solid var(--line);border-radius:7px}.owg-period [data-period-summary]{flex-basis:100%;font-size:12px;color:var(--muted)}.owg-period [data-custom]{display:flex;gap:8px;align-items:center;flex-wrap:wrap}.owg-period [hidden]{display:none!important}';document.head.appendChild(style);
+    const today=new Date();start=dateInput(new Date(today.getFullYear(),today.getMonth(),today.getDate()-6));end=dateInput(today);
+    for(const id of ['overview','evidence','connect','export']) {
+      const panel=$('#panel-'+id);if(!panel)continue;
+      const bar=document.createElement('div');bar.className='owg-period';
+      bar.innerHTML=`Work period${esc(Intl.DateTimeFormat().resolvedOptions().timeZone||'local time')}`;
+      // Scope applies to handoff/exports; existing current-run dashboards retain their labels.
+      const note=document.createElement('small');note.className='muted';note.textContent='Select the period to view in Activity or prepare for AI. Activity uses this period; current-run charts keep their labeled scope.';bar.appendChild(note);
+      panel.insertBefore(bar,panel.firstChild);
+      bar.querySelector('select').onchange=e=>{period=e.target.value;syncControls();refresh();document.dispatchEvent(new Event('owg:period-change'));};
+      bar.querySelectorAll('input').forEach(x=>x.onchange=()=>{start=bar.querySelector('[data-start]').value;end=bar.querySelector('[data-end]').value;syncControls();refresh();document.dispatchEvent(new Event('owg:period-change'));});
+      bar.querySelector('button').onclick=prepare;
+    }
+    syncControls();refresh();document.dispatchEvent(new Event('owg:period-change'));
+  }
+  if(document.readyState==='loading')document.addEventListener('DOMContentLoaded',install);else install();
+})();
diff --git a/dashboard/basic_mode.js b/dashboard/basic_mode.js
index 7db5bd24..ff9974ee 100644
--- a/dashboard/basic_mode.js
+++ b/dashboard/basic_mode.js
@@ -564,6 +564,18 @@
       title.textContent = 'Recording is paused'; sub.textContent = 'Nothing is recorded until you resume.'; dot.className = 'hero-dot paused';
     } else if (state === 'stopped') {
       title.textContent = 'Recording is stopped'; sub.textContent = 'Your data is kept on this computer.'; dot.className = 'hero-dot off';
+    } else if (capture.missing_permissions?.length) {
+      title.textContent = 'Recorder needs permission';
+      sub.textContent = 'OpenWorkGraph is waiting for the required OS permissions. Saved work remains available.';
+      dot.className = 'hero-dot paused';
+    } else if (capture.away) {
+      title.textContent = 'Recorder is away';
+      sub.textContent = 'The computer is idle or locked. Saved work remains available.';
+      dot.className = 'hero-dot paused';
+    } else if (capture.collector_alive === false) {
+      title.textContent = 'Waiting for the recorder';
+      sub.textContent = 'No collector heartbeat has arrived for this run. Saved work remains available.';
+      dot.className = 'hero-dot paused';
     } else {
       title.textContent = engaged >= 60 ? `${fmtDuration(engaged)} of work recorded so far` : 'Recording your work';
       const parts = [];
diff --git a/dashboard/evidence_paging.js b/dashboard/evidence_paging.js
index 2d90a1d9..778ef98f 100644
--- a/dashboard/evidence_paging.js
+++ b/dashboard/evidence_paging.js
@@ -25,7 +25,7 @@
   }
 
   function endpoint(cursor) {
-    const params=new URLSearchParams({scope:'current',limit:String(PAGE_SIZE)});
+    const params=new URLSearchParams({...(window.owgEvidenceSelection?.()||{scope:'current'}),limit:String(PAGE_SIZE)});
     const q=currentQuery();
     if (q) params.set('q',q);
     if (selectedSurface) params.set('surface',selectedSurface);
@@ -168,6 +168,7 @@
     document.querySelectorAll('[data-evidence-view]').forEach(button=>button.addEventListener('click',()=>{
       if (button.dataset.evidenceView==='all') schedulePageLoad(0);
     }));
+    document.addEventListener('owg:period-change',()=>{resetPages();loadPage();});
     ensureControls();
     if (window.__owgLastSummary) pagedRenderEvidence(window.__owgLastSummary);
   }
diff --git a/dashboard/gateway_panel.js b/dashboard/gateway_panel.js
index d143849c..4f9a00ff 100644
--- a/dashboard/gateway_panel.js
+++ b/dashboard/gateway_panel.js
@@ -323,7 +323,7 @@
       const runs=(p.runs||[]).map(run=>`${e(String(run.started_at||'').replace('T',' ').slice(0,16))}–${e(String(run.ended_at||'').slice(11,16))}${(run.surfaces||[]).map(e).join(' → ')}${fmtDuration(run.engaged_seconds)}${e(String((run.boundary||{}).end_reason||'').replace(/_/g,' '))}`).join('');
       const data=encodeURIComponent(JSON.stringify({name:p.name,signature:p.signature,event_ids:(p.runs||[]).flatMap(r=>r.event_ids||[])}));
       return `
${e(safeName)} Needs review · ${e(p.confidence||'low')}
${steps}
×${Number(p.observed_count||p.runs?.length||0)}
median ${fmtDuration(p.median_engaged_seconds)} · total ${fmtDuration(p.total_engaged_seconds)}
${runs||''}
WhenStepsEngaged timeHow it ended
Execution windows are unavailable for this pattern.
`;
-    }).join('') || (typeof window.emptyState==='function' ? window.emptyState('No repeated workflow yet','When you do the same sequence of steps across tools at least twice, it appears here with how long each run took.') : '
No repeated workflow yet.
');
+    }).join('') || (typeof window.emptyState==='function' ? window.emptyState('No repeated pattern found in this run','You can still prepare the recorded chronology for AI analysis.') : '
No repeated pattern found in this run. You can still prepare the recorded chronology for AI analysis.
');
     host.querySelectorAll('[data-show-runs]').forEach(btn=>btn.onclick=()=>document.querySelector(`#rich-pattern-${btn.dataset.showRuns}`)?.classList.toggle('expanded'));
     host.querySelectorAll('[data-automation]').forEach(btn=>btn.onclick=()=>{if(typeof window.activateTab==='function')window.activateTab('connect');if(typeof window.toast==='function')window.toast('Connect your AI, then ask it to inspect this repeated workflow through OpenWorkGraph MCP.');});
     host.querySelectorAll('[data-pattern-export]').forEach(btn=>btn.onclick=()=>{
diff --git a/dashboard/index.html b/dashboard/index.html
index f1a467f1..5937585b 100644
--- a/dashboard/index.html
+++ b/dashboard/index.html
@@ -130,13 +130,13 @@
         
—
Current-run pattern runs

         
—
Key presses

       
-      
Today's timeline
One lane per tool, built from the time each window or site was in front. Browser time is attributed to the site you were on (Gmail, Salesforce) rather than to the browser.
Loading your timeline…

+      
Current run timeline
One lane per tool, built from the time each window or site was in front. Browser time is attributed to the site you were on (Gmail, Salesforce) rather than to the browser.
Loading your timeline…

       
Patterns in this run
This card counts only the current recording run; saved-history workflow examples appear separately below. Both are derived and may disagree until sufficient activity is observed.

-      
Time by tool
Engaged time (actively working, not just open) per tool. Browser time is attributed to the site when it is known.

+      
Time by tool
Estimated engaged time (recent input, not proof of continuous work) per tool. Browser time is attributed to the site when it is known.

     
 
     
 
     
   
   

@@ -203,7 +203,7 @@ 
ChatGPT cloud sharing (optional)

 function updateTabScrollHint(){const tabs=document.querySelector('.tabs');if(!tabs)return;const max=tabs.scrollWidth-tabs.clientWidth;tabs.classList.toggle('more-left',max>2&&tabs.scrollLeft>2);tabs.classList.toggle('more-right',max>2&&tabs.scrollLeftbutton.textContent=old,1400);}}catch(_){window.prompt('Copy this:',text);}}
 function b64(s){return btoa(unescape(encodeURIComponent(s))).replace(/\+/g,'-').replace(/\//g,'_').replace(/=+$/,'');}
 function connectCursor(){const config=b64(JSON.stringify({url:MCP_ENDPOINT}));window.location.href=`cursor://anysphere.cursor-deeplink/mcp/install?name=OpenWorkGraph&config=${encodeURIComponent(config)}`;}
diff --git a/docs/EXPORTS_AND_AI.md b/docs/EXPORTS_AND_AI.md
index d9bb8d39..e52d4503 100644
--- a/docs/EXPORTS_AND_AI.md
+++ b/docs/EXPORTS_AND_AI.md
@@ -110,7 +110,7 @@ For more conservative analysis, also ask the model to state what evidence would
 
 ## AI access is explicit
 
-MCP reads are **OFF on every OpenWorkGraph launch**.
+New installations default AI access **ON at Redacted**. An explicit user opt-out or reset-on-restart preference is preserved; recording remains a separate choice.
 
 The dashboard's **AI access** switch controls whether MCP tools may read workflow evidence. The server checks that switch on every tool call, so turning access off also blocks an AI client that was already configured or connected.
 
@@ -258,7 +258,7 @@ An export is a file the user can inspect before uploading.
 
 With MCP, an AI client can decide which tool to call after access has been enabled. **Evidence returned by an MCP call is sent to the AI provider so that model can process it.** Keeping the OpenWorkGraph database local does not mean the returned tool data stays local to the computer.
 
-The AI-access switch (off on a new install; remembered after a restart unless you choose otherwise) and the local activity view exist to make that boundary explicit.
+The AI-access switch (on at Redacted on a new install; remembered after a restart unless you choose otherwise) and the local activity view exist to make that boundary explicit.
 
 ---
 
@@ -306,3 +306,53 @@ GET      /v1/mcp-connection-config
 ```
 
 All data/control endpoints require the appropriate authenticated local session/capability in normal operation.
+
+## Selected-period AI handoff
+
+Basic mode now has a **Work period** selector on Overview/Today, Activity, AI apps,
+and Export. Choose Today, Last 7 days, Custom dates, This run or All retained work.
+Activity paging and ordinary exports follow this selection. Current-run charts
+remain explicitly labeled as current-run views.
+
+**Prepare for AI** previews the selected and observed dates, event/source counts,
+redacted context examples, local saved-history permission and missing-evidence
+limits. It can copy an opportunity-discovery request or download a redacted CSV
+ZIP with the existing guide, prompt and manifest. It does not invoke a model,
+upload a file, enable recording or grant AI access. No repeated-work candidate or
+upfront workflow description is required. Detailed event evidence is an optional
+choice, separate from visible work-text capture (which is not in these exports).
+
+Custom calendar days use the browser's local timezone and are converted to UTC,
+including daylight-saving changes. The end date is inclusive in the picker;
+the wire-level end timestamp is exclusive.
+
+The existing JSON/XLSX/CSV ZIP route also accepts a bounded selection:
+
+```text
+GET /v1/export/csvzip?scope=range&since=2026-10-12T00:00:00Z&until=2026-10-17T00:00:00Z&include_raw=true&redact_names=true
+```
+
+Both bounds must include a timezone, and end must be after start. Bounds require
+`scope=range`. Existing `scope=current` and `scope=all` calls retain their defaults.
+Download tickets bind the dates as well as format, scope, detail and redaction.
+The same bounds apply to Activity cursor fingerprints and all exported layers.
+Canonical observations are selected by their start timestamp in `[since, until)`;
+original span durations remain intact in canonical evidence. Derived timing uses
+only the portion before the selected end, with proportional aggregate input
+estimates. An observation starting before the window is not fetched implicitly.
+`CAPTURE_MANIFEST.json.window` documents those semantics and actual observed bounds.
+
+Event CSVs append `event_id` after their existing columns, allowing an AI to cite
+specific observations without changing existing column positions. Factual context,
+tasks and work profiles remain derived views, separate from canonical evidence.
+A small or empty repeated-work index does not establish absence of automatable work.
+
+`GET /v1/analysis-preview` is a **dashboard-session-only** preparation endpoint.
+It is not an MCP history-access bypass. Local history availability is not proof
+that a hosted app has received it; hosted availability follows that deployment's
+existing sharing/enrollment boundary. Preparing or downloading an evidence package
+never changes either permission.
+
+On upgrade, the checkpointed title-protection migration revisits retained rows
+to apply the stronger person-search protection and rebuild the derived views.
+Already-tokenized titles and the remaining business context are preserved.
diff --git a/server/ai_context.py b/server/ai_context.py
index 9bfd0d9d..0e77750b 100644
--- a/server/ai_context.py
+++ b/server/ai_context.py
@@ -291,7 +291,11 @@ def learned_token(name: str) -> str | None:
 
 def contextual_text_redactor(settings: dict[str, Any] | None = None):
     settings = settings or user_settings()
-    return _REDACTOR.get(_detector_state(settings), lambda: _build_detector(settings))
+    from .contextual_redaction_extensions import augment
+    return augment(_REDACTOR.get(_detector_state(settings), lambda: _build_detector(settings)), settings)
+
+
+_DOGFOOD_REDACTION_EXTENSIONS = True  # factory includes extensions without HTTP import side effects
 
 
 def _redact_slug_component(component: str, redact) -> str:
diff --git a/server/analysis_handoff.py b/server/analysis_handoff.py
new file mode 100644
index 00000000..5738acde
--- /dev/null
+++ b/server/analysis_handoff.py
@@ -0,0 +1,62 @@
+"""Local dashboard preview for a user-selected evidence handoff. No AI invocation."""
+from __future__ import annotations
+
+from fastapi import HTTPException, Request
+from .export_window import selection, sql_bounds
+from .exporter import _analysis_request
+from .db import connect
+
+
+def analysis_preview(request: Request, scope: str = "current", since: str | None = None,
+                     until: str | None = None) -> dict:
+    # A preparation screen is not another way for an AI to read ungranted history.
+    from .secure_app import _dashboard_authenticated
+    if not _dashboard_authenticated(request):
+        raise HTTPException(status_code=403, detail="Local dashboard session required")
+    try:
+        window = selection(scope, since, until)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from None
+    clause, params = sql_bounds(window["since"], window["until"])
+    where = " WHERE " + clause if clause else ""
+    with connect() as conn:
+        grouped = conn.execute("SELECT event_type, source, COUNT(*) AS count FROM events" + where
+                               + " GROUP BY event_type, source", params).fetchall()
+        available = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
+        samples = conn.execute("SELECT event_id, observed_at, app, window_title FROM events" + where
+                               + (" AND " if where else " WHERE ") + "event_type='focus_span'"
+                               + " ORDER BY julianday(observed_at) DESC LIMIT 6", params).fetchall()
+        bounds = conn.execute("SELECT observed_at FROM events" + where
+                              + " ORDER BY julianday(observed_at), id LIMIT 1", params).fetchone()
+        last = conn.execute("SELECT observed_at FROM events" + where
+                            + " ORDER BY julianday(observed_at) DESC, id DESC LIMIT 1", params).fetchone()
+    from .ai_context import redact_contextually
+    from shared.history_policy import active_ai_history_access
+    from .work_text_capture import get_policy
+    from .ai_access import ai_access_enabled
+    counts: dict[str, int] = {}
+    sources: dict[str, int] = {}
+    for row in grouped:
+        counts[row["event_type"]] = counts.get(row["event_type"], 0) + row["count"]
+        sources[row["source"]] = sources.get(row["source"], 0) + row["count"]
+    return {
+        "window": window,
+        "first_observed_at": bounds[0] if bounds else None,
+        "last_observed_at": last[0] if last else None,
+        "canonical_rows": sum(counts.values()), "event_types": counts,
+        "sources": sources, "retained_rows": available,
+        "redacted_examples": redact_contextually([dict(x) for x in samples]),
+        "local_ai_enabled": ai_access_enabled(),
+        "local_ai_history_access": active_ai_history_access(),
+        "work_text": {"included_in_activity_package": False,
+                      "local_capture_enabled": bool(get_policy().get("capture_enabled"))},
+        "request_text": _analysis_request(window) + "\n\nUse the connected OWG chronology tools, without a search filter, "
+                        "for this period and follow their pagination. Respect the currently granted history boundary. "
+                        "If these dates are unavailable, explain the missing access rather than widening it. "
+                        "An attached evidence package can also be used without MCP.",
+        "destination": "Nothing sent. Choose a file download or copy a request for your AI.",
+        "hosted_history_availability": "Not established by this local preview; hosted apps only see evidence already shared with them.",
+        "limitations": ["Gaps are unobserved intervals, not lost work.",
+                        "Observed dates and event counts do not establish complete coverage.",
+                        "No repeated-work candidates are required to analyze the chronology."],
+    }
diff --git a/server/analytics.py b/server/analytics.py
index 781f29d0..a1921248 100644
--- a/server/analytics.py
+++ b/server/analytics.py
@@ -905,9 +905,9 @@ def close(end_ts: float, end_reason: str, next_start_reason: str) -> None:
         },
     }
 
-def semantic_activity(limit: int = 200, since: str | None = None, *, operational: bool = False) -> list[dict[str, Any]]:
+def semantic_activity(limit: int = 200, since: str | None = None, *, operational: bool = False, _raw_events: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
     loader = _operational_event_rows if operational else _event_rows
-    events = loader(max(limit * 6, 500), since=since)
+    events = _raw_events if _raw_events is not None else loader(max(limit * 6, 500), since=since)
     semantic: list[dict[str, Any]] = []
     for e in events:
         et = str(e.get("event_type") or "")
@@ -940,8 +940,8 @@ def semantic_activity(limit: int = 200, since: str | None = None, *, operational
     return semantic[-max(1, min(limit, 2000)):][::-1]
 
 
-def _summary_uncached(limit: int = 10000, since: str | None = None, *, operational: bool = False) -> dict[str, Any]:
-    events = _operational_event_rows(limit, since=since) if operational else _event_rows(limit, since=since)
+def _summary_uncached(limit: int = 10000, since: str | None = None, *, operational: bool = False, _raw_events: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+    events = _raw_events if _raw_events is not None else (_operational_event_rows(limit, since=since) if operational else _event_rows(limit, since=since))
     focus_events = [e for e in events if e.get("event_type", "focus_span") == "focus_span"]
     interaction_events = [e for e in events if str(e.get("event_type", "")).startswith("screen_")]
     browser_events = [e for e in events if str(e.get("event_type", "")).startswith("browser_")]
@@ -1124,7 +1124,7 @@ def _summary_uncached(limit: int = 10000, since: str | None = None, *, operation
     task_data = candidate_tasks(
         limit=limit,
         since=since,
-        _raw_events=events if not operational else None,
+        _raw_events=events if _raw_events is not None or not operational else None,
     )
 
     return {
diff --git a/server/basic_mode_routes.py b/server/basic_mode_routes.py
index d7c429ff..3a14589e 100644
--- a/server/basic_mode_routes.py
+++ b/server/basic_mode_routes.py
@@ -49,7 +49,7 @@ async def _inject_basic_mode_script(request: Request, call_next):
         text = body.decode("utf-8")
     except Exception:
         return response
-    marker = ''
+    marker = '\n'
     if marker not in text:
         text = text.replace("", marker + "\n")
     headers = dict(response.headers)
@@ -62,6 +62,11 @@ async def _inject_basic_mode_script(request: Request, call_next):
     app.add_api_route("/v1/capture-exclusions", put_capture_exclusions, methods=["PUT"])
     app.add_api_route("/basic-mode.js", basic_mode_script, methods=["GET"])
     app.add_api_route("/v1/update-status", update_status, methods=["GET"])
+    from .analysis_handoff import analysis_preview
+    app.add_api_route("/v1/analysis-preview", analysis_preview, methods=["GET"])
+    app.add_api_route("/analysis-handoff.js", lambda: Response(
+        (ROOT / "dashboard" / "analysis_handoff.js").read_text(encoding="utf-8"),
+        media_type="application/javascript"), methods=["GET"])
     app.middleware("http")(_inject_basic_mode_script)
     app.state.owg_basic_mode_installed = True
 
diff --git a/server/context_exporter.py b/server/context_exporter.py
index 582502b4..f15e37d7 100644
--- a/server/context_exporter.py
+++ b/server/context_exporter.py
@@ -36,11 +36,21 @@
 ]
 
 
-def build_export_payload(*, scope: str = "current", include_raw: bool = False) -> dict[str, Any]:
-    payload = base.build_export_payload(scope=scope, include_raw=include_raw)
-    since = base._since_for_scope(scope)
-    task_data = candidate_tasks(limit=100000, since=since)
-    profile = compute_work_profile(scope=scope)
+def build_export_payload(*, scope: str = "current", include_raw: bool = False, since: str | None = None, until: str | None = None) -> dict[str, Any]:
+    # Validate even when called directly rather than through the HTTP route.
+    base.selection(scope, since, until)
+    if scope == "range":
+        payload = base.build_export_payload(scope=scope, include_raw=include_raw, since=since, until=until)
+        window = payload["capture_manifest"]["window"]
+        since, until = window["since"], window["until"]
+        evidence = base._query_table("events", since=since, until=until)
+        task_data = candidate_tasks(_raw_events=base.analysis_rows(evidence, until))
+        profile = compute_work_profile(scope=scope, since=since, until=until)
+    else:
+        payload = base.build_export_payload(scope=scope, include_raw=include_raw)
+        since = base._since_for_scope(scope)
+        task_data = candidate_tasks(limit=100000, since=since)
+        profile = compute_work_profile(scope=scope)
 
     payload["inferred_tasks"] = task_data.get("tasks", [])
     payload["repeated_task_families"] = task_data.get("patterns", [])
diff --git a/server/contextual_redaction_extensions.py b/server/contextual_redaction_extensions.py
index 365b78f0..4640a870 100644
--- a/server/contextual_redaction_extensions.py
+++ b/server/contextual_redaction_extensions.py
@@ -13,7 +13,7 @@
 from typing import Any, Callable
 
 from . import presentation
-from .contextual_redaction import AMBIGUOUS_NAMES, FIRST_NAMES, SURNAMES
+from .contextual_redaction import AMBIGUOUS_NAMES, FIRST_NAMES, SURNAMES, STOPWORDS, ORGANIZATIONS, PLACE_ORG_NOUNS
 
 _WORD = r"[A-Za-zÀ-ÖØ-öø-ÿ][A-Za-zÀ-ÖØ-öø-ÿ'’-]*"
 _TOKEN = r"(?:OWNER|PERSON(?:_[0-9A-F]{4,})?|[A-Z][A-Z_]*_[0-9A-F]{4,})"
@@ -25,6 +25,7 @@
 _TOKEN_SLASH_AFTER_RE = re.compile(rf"\b({_TOKEN})\s*/\s*({_WORD})\b")
 _TOKEN_SLASH_BEFORE_RE = re.compile(rf"\b({_WORD})\s*/\s*({_TOKEN})\b")
 _SEGMENT_SPLIT_RE = re.compile(r"\s+[-–—|·]\s+|\s*[|·]\s*|:\s+")
+_SEARCH_TITLE_RE = re.compile(r"\s[-–—]\s(?:Google Search|Sök på Google)(?:\s[-–—].*)?$", re.I)
 _STRONG_CUE_RE = re.compile(
     r"(?:chat|meeting|meet|call|message|email|mail|reply|cc|bcc|attendee|assigned|assignee|owner|contact|"
     r"dear|hi|hello|hej|möte|samtal|ring|patient|candidate)\s+(?:with|from|to|med|från)?\s*$",
@@ -34,6 +35,11 @@
 _FIRST = frozenset(name.casefold() for name in FIRST_NAMES)
 _LAST = frozenset(name.casefold() for name in SURNAMES)
 _AMBIG = frozenset(name.casefold() for name in AMBIGUOUS_NAMES)
+_SEARCH_NON_NAMES = frozenset(x.casefold() for x in STOPWORDS | ORGANIZATIONS | PLACE_ORG_NOUNS) | {
+    "price", "pricing", "weather", "invoice", "template", "calculator", "cardiology",
+    "university", "hospital", "dress", "shoes", "software", "login", "support",
+    "jobs", "careers", "recipe", "reviews", "meaning", "translation",
+}
 
 
 def _base(value: str) -> str:
@@ -92,7 +98,9 @@ def particle(match: re.Match[str]) -> str:
         fk, lk = _base(first), _base(last)
         if first != first.lower() or last != last.lower():
             continue
-        if fk not in _FIRST or lk not in _LAST or fk in _AMBIG or lk in _AMBIG:
+        search_lead = match.start() == 0 and bool(_SEARCH_TITLE_RE.search(out[match.end():]))
+        search_surname = search_lead and lk not in _SEARCH_NON_NAMES and len(lk) >= 3
+        if fk not in _FIRST or (lk not in _LAST and not search_surname) or fk in _AMBIG or lk in _AMBIG:
             continue
         seg_exact = any(
             out[a:b].strip(" ,;()[]") == match.group(0)
@@ -100,7 +108,10 @@ def particle(match: re.Match[str]) -> str:
             if a <= match.start() and match.end() <= b
         )
         cue = bool(_STRONG_CUE_RE.search(out[:match.start()]))
-        if not (seg_exact or cue) or _in_never(out, match.start(), match.end(), never):
+        # A non-ambiguous given name leading a person-shaped search can have
+        # an uncommon surname absent from the bundled list. Preserve qualifiers.
+        # Ordinary product/business query terms above remain untouched.
+        if not (seg_exact or cue or search_lead) or _in_never(out, match.start(), match.end(), never):
             continue
         start, end = match.start() + offset, match.end() + offset
         replacement = _token(match.group(0))
diff --git a/server/db.py b/server/db.py
index 3a8675aa..9b43594e 100644
--- a/server/db.py
+++ b/server/db.py
@@ -4,6 +4,9 @@
 import json
 import os
 import sqlite3
+import threading
+import time
+import logging
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any, Iterable
@@ -359,7 +362,9 @@ def harden_existing_browser_events(config: dict[str, Any]) -> int:
     return changed
 
 
-TITLE_PROTECTION_MIGRATION = "titles_sensitive_details_tokenized_v0108"
+# Revisit existing installations too: the first pass could miss qualified
+# lowercase person searches. Keep the same checkpointed, idempotent migration.
+TITLE_PROTECTION_MIGRATION = "titles_sensitive_details_tokenized_v0108_search_names_v2"
 
 
 def _protected_text_changed(before: dict[str, Any], after: dict[str, Any]) -> bool:
@@ -374,15 +379,15 @@ def texts(event: dict[str, Any]) -> tuple:
 
 
 _TITLE_PROTECTION_DONE: set[str] = set()
-SYNC_TITLE_PROTECTION_ROWS = 5000
+_TITLE_PROTECTION_WORKERS: set[str] = set()
+_TITLE_PROTECTION_WORKERS_LOCK = threading.Lock()
 
 
 def title_protection_complete() -> bool:
-    """True once every row stored before v0.108 has had its titles protected.
+    """True only after the entire checkpointed historical re-protection finishes.
 
-    Until then older rows may still hold names, so callers that would otherwise
-    return stored text unredacted (Full AI context, unredacted exports) must
-    redact on read instead. Cached once true (the marker is never removed).
+    Until then Full AI context and unredacted exports must fail closed to the
+    redacted representation, including while a background worker is sleeping.
     """
     key = str(DB_PATH)
     if key in _TITLE_PROTECTION_DONE:
@@ -390,7 +395,8 @@ def title_protection_complete() -> bool:
     try:
         with connect() as conn:
             done = bool(conn.execute(
-                "SELECT 1 FROM privacy_migrations WHERE migration_key = ?", (TITLE_PROTECTION_MIGRATION,)
+                "SELECT 1 FROM privacy_migrations WHERE migration_key = ?",
+                (TITLE_PROTECTION_MIGRATION,),
             ).fetchone())
     except Exception:
         return False
@@ -399,44 +405,75 @@ def title_protection_complete() -> bool:
     return done
 
 
-def protect_existing_titles_in_background() -> None:
-    """Run the one-time title migration without delaying startup.
+def mark_title_protection_complete_if_empty() -> bool:
+    """Constant-time fresh-install fast path; never scan retained work at startup.
 
-    Until it finishes (``title_protection_complete``), AI context is forced to
-    Redacted and exports are forced to redact on read, so older rows are never
-    returned with the personal details the migration removes.
+    A genuinely empty canonical store contains nothing to re-protect. Newly
+    collected events still go through the current capture-time title policy.
+    Existing histories, even a single row, are always deferred so no upgrade
+    workload can block the launcher health probe.
     """
-    import threading
-
     if title_protection_complete():
-        return
-    # Small histories (fresh installs, most upgrades) finish in well under a
-    # second: do them now so Full context is never briefly unavailable. Only a
-    # long history goes to the background.
-    try:
-        with connect() as conn:
-            conn.execute("CREATE TABLE IF NOT EXISTS privacy_migration_progress (migration_key TEXT PRIMARY KEY, last_id INTEGER NOT NULL)")
-            row = conn.execute("SELECT last_id FROM privacy_migration_progress WHERE migration_key = ?", (TITLE_PROTECTION_MIGRATION,)).fetchone()
-            remaining = int(conn.execute("SELECT COUNT(*) FROM events WHERE id > ? AND source != 'agent'", (int(row[0]) if row else 0,)).fetchone()[0])
-    except Exception:
-        remaining = SYNC_TITLE_PROTECTION_ROWS + 1
-    if remaining <= SYNC_TITLE_PROTECTION_ROWS:
-        try:
-            protect_existing_titles()
+        return True
+    with connect() as conn:
+        if conn.execute("SELECT 1 FROM events LIMIT 1").fetchone():
+            return False
+        conn.execute(
+            "INSERT OR IGNORE INTO privacy_migrations(migration_key) VALUES (?)",
+            (TITLE_PROTECTION_MIGRATION,),
+        )
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS privacy_migration_progress "
+            "(migration_key TEXT PRIMARY KEY, last_id INTEGER NOT NULL)",
+        )
+        conn.execute(
+            "DELETE FROM privacy_migration_progress WHERE migration_key = ?",
+            (TITLE_PROTECTION_MIGRATION,),
+        )
+    _TITLE_PROTECTION_DONE.add(str(DB_PATH))
+    return True
+
+
+def protect_existing_titles_in_background() -> None:
+    """Spawn a strictly asynchronous, throttled, checkpointed upgrade worker.
+
+    Called *after* the ASGI lifespan has yielded and the app has had time to
+    start serving health checks. Even a 5k-row history is not migrated on the
+    request/event-loop thread. One worker per DB per process prevents races.
+    """
+    key = str(DB_PATH)
+    with _TITLE_PROTECTION_WORKERS_LOCK:
+        if key in _TITLE_PROTECTION_WORKERS:
             return
-        except Exception:
-            pass  # fall through to the background attempt
+        _TITLE_PROTECTION_WORKERS.add(key)
 
     def run() -> None:
         try:
-            protect_existing_titles()
-        except Exception:
-            pass  # retried on the next start (progress is checkpointed)
+            if not title_protection_complete():
+                # Short transactions + a pause after each committed checkpoint
+                # leave I/O and CPU headroom for health checks and live capture.
+                protect_existing_titles(batch_size=200, pause_seconds=0.02)
+        except Exception as exc:
+            # Never print exception values or historical row contents.
+            logging.getLogger(__name__).warning(
+                "Deferred title-protection migration will resume after restart (%s)",
+                type(exc).__name__,
+            )
+        finally:
+            with _TITLE_PROTECTION_WORKERS_LOCK:
+                _TITLE_PROTECTION_WORKERS.discard(key)
 
-    threading.Thread(target=run, name="owg-title-protection-migration", daemon=True).start()
+    try:
+        threading.Thread(
+            target=run, name="owg-title-protection-migration", daemon=True,
+        ).start()
+    except BaseException:
+        with _TITLE_PROTECTION_WORKERS_LOCK:
+            _TITLE_PROTECTION_WORKERS.discard(key)
+        raise
 
 
-def protect_existing_titles(*, batch_size: int = 1000) -> int:
+def protect_existing_titles(*, batch_size: int = 1000, pause_seconds: float = 0.0) -> int:
     """Apply the v0.108 title rule once to evidence stored before it.
 
     Earlier versions stored desktop-app titles, control labels and URL paths with
@@ -486,6 +523,11 @@ def protect_existing_titles(*, batch_size: int = 1000) -> int:
                 "INSERT OR REPLACE INTO privacy_migration_progress(migration_key, last_id) VALUES (?, ?)",
                 (TITLE_PROTECTION_MIGRATION, last_id),
             )
+        # Outside the SQLite transaction: readers, writes from the capture
+        # worker, and the single-threaded API remain responsive during an
+        # upgrade of hundreds of thousands of historical rows.
+        if pause_seconds > 0:
+            time.sleep(min(float(pause_seconds), 0.2))
 
 
 def insert_events(events: Iterable[dict[str, Any]]) -> int:
diff --git a/server/evidence_paging.py b/server/evidence_paging.py
index fdfcb436..f253bd82 100644
--- a/server/evidence_paging.py
+++ b/server/evidence_paging.py
@@ -28,14 +28,18 @@ def get_paged_evidence(
     cursor: str | None = None,
     surface: str | None = None,
     q: str | None = None,
+    since: str | None = None, until: str | None = None,
 ) -> dict[str, Any]:
-    if scope not in {"current", "all"}:
-        raise HTTPException(status_code=400, detail="scope must be current or all")
-    since = enterprise_app._current_run_start() if scope == "current" else None
+    from .export_window import selection
     try:
+        window = selection(scope, since, until)
+        since, until = window["since"], window["until"]
+        if scope == "current":
+            since = enterprise_app._current_run_start()
         return query_evidence(
             scope=scope,
             since=since,
+            until=until,
             limit=limit,
             cursor=cursor,
             surface=surface,
diff --git a/server/evidence_query.py b/server/evidence_query.py
index 8aca5d02..72ac4457 100644
--- a/server/evidence_query.py
+++ b/server/evidence_query.py
@@ -20,13 +20,14 @@ def _norm_filter(value: str | None, limit: int) -> str:
     return " ".join(str(value or "").strip().split())[:limit]
 
 
-def _fingerprint(scope: str, surface: str, query: str, since: str | None, *, include_agents: bool = False) -> str:
+def _fingerprint(scope: str, surface: str, query: str, since: str | None, *, include_agents: bool = False, until: str | None = None) -> str:
     raw = json.dumps(
         {
             "scope": scope,
             "surface": surface.casefold(),
             "q": query.casefold(),
             "since": str(since or ""),
+            "until": str(until or ""),
             "include_agents": bool(include_agents),
         },
         sort_keys=True,
@@ -92,6 +93,7 @@ def _where(
     query: str,
     include_surface: bool = True,
     include_agents: bool = False,
+    until: str | None = None,
 ) -> tuple[list[str], list[Any]]:
     clauses: list[str] = []
     params: list[Any] = []
@@ -99,8 +101,11 @@ def _where(
         # Agent runs have their own tab; this is the human evidence view.
         clauses.append("source != 'agent'")
     if since:
-        clauses.append("observed_at >= ?")
+        clauses.append("julianday(observed_at) >= julianday(?)")
         params.append(str(since))
+    if until:
+        clauses.append("julianday(observed_at) < julianday(?)")
+        params.append(until)
     if include_surface and surface:
         clauses.append("LOWER(surface) = LOWER(?)")
         params.append(surface)
@@ -178,14 +183,19 @@ def query_evidence(
     *,
     scope: str,
     since: str | None,
+    until: str | None = None,
     limit: int = 100,
     cursor: str | None = None,
     surface: str | None = None,
     q: str | None = None,
     include_agents: bool = False,
 ) -> dict[str, Any]:
-    if scope not in {"current", "all"}:
-        raise ValueError("scope must be current or all")
+    if scope not in {"current", "all", "range"}:
+        raise ValueError("scope must be current, all, or range")
+    if scope == "range":
+        from .export_window import selection
+        window = selection(scope, since, until)
+        since, until = window["since"], window["until"]
     page_limit = max(1, min(int(limit), _MAX_LIMIT))
     selected_surface = _norm_filter(surface, _MAX_SURFACE)
     query = _norm_filter(q, _MAX_QUERY)
@@ -195,6 +205,7 @@ def query_evidence(
         query,
         since,
         include_agents=include_agents,
+        until=until,
     )
     cursor_ts = ""
     cursor_id = 0
@@ -208,18 +219,19 @@ def query_evidence(
         query=query,
         include_surface=True,
         include_agents=include_agents,
+        until=until,
     )
     page_clauses = list(base_clauses)
     page_params = list(base_params)
     if cursor_ts:
-        page_clauses.append("(observed_at < ? OR (observed_at = ? AND id < ?))")
+        page_clauses.append("(julianday(observed_at) < julianday(?) OR (julianday(observed_at) = julianday(?) AND id < ?))")
         page_params.extend([cursor_ts, cursor_ts, cursor_id])
 
     select_sql = (
         "SELECT id, event_id, observed_at, source, surface, action, resource_title, "
         "resource_locator, target_label, metadata_json FROM context_events"
         + _sql_where(page_clauses)
-        + " ORDER BY observed_at DESC, id DESC LIMIT ?"
+        + " ORDER BY julianday(observed_at) DESC, id DESC LIMIT ?"
     )
 
     facet_clauses, facet_params = _where(
@@ -228,6 +240,7 @@ def query_evidence(
         query=query,
         include_surface=False,
         include_agents=include_agents,
+        until=until,
     )
 
     with db.connect() as conn:
@@ -262,6 +275,7 @@ def query_evidence(
     return {
         "scope": scope,
         "since": since,
+        "until": until,
         "surface": selected_surface or None,
         "q": query,
         "limit": page_limit,
diff --git a/server/export_window.py b/server/export_window.py
new file mode 100644
index 00000000..0b423593
--- /dev/null
+++ b/server/export_window.py
@@ -0,0 +1,75 @@
+"""Validated export selection, shared by previews, downloads and download tickets."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+import copy
+import os
+
+
+def timestamp(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (ValueError, TypeError):
+        raise ValueError("Use an ISO date/time with a timezone") from None
+    if parsed.tzinfo is None:
+        raise ValueError("Date/time must include a timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+def observed_timestamp(value: str) -> datetime:
+    # Historical rows may predate timezone-qualified capture. SQLite and the
+    # existing analytics interpret those as UTC; keep that legacy convention.
+    parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    return (parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)).astimezone(timezone.utc)
+
+
+def selection(scope: str = "current", since: str | None = None,
+              until: str | None = None) -> dict[str, Any]:
+    if scope not in {"current", "all", "range"}:
+        raise ValueError("scope must be current, all, or range")
+    if scope == "range":
+        if not since or not until:
+            raise ValueError("Choose both start and end dates")
+        start, end = timestamp(since), timestamp(until)
+        if start >= end:
+            raise ValueError("End must be after start")
+        since, until = start.isoformat(), end.isoformat()
+    elif since is not None or until is not None:
+        raise ValueError("Date bounds require scope=range")
+    else:
+        since = os.getenv("WORKFLOW_OBSERVER_RUN_STARTED_AT") if scope == "current" else None
+    return {"scope": scope, "since": since, "until": until,
+            "bounds": "start_inclusive_end_exclusive", "timezone": "UTC",
+            "event_selection": "observed_at_in_window"}
+
+
+def sql_bounds(since: str | None, until: str | None, column: str = "observed_at"):
+    # Column is supplied by code, never a request parameter.
+    parts, params = [], []
+    for bound, op in ((since, ">="), (until, "<")):
+        if bound:
+            parts.append(f"julianday({column}) {op} julianday(?)")
+            params.append(bound)
+    return " AND ".join(parts), tuple(params)
+
+
+def analysis_rows(events: list[dict[str, Any]], until: str | None) -> list[dict[str, Any]]:
+    """Clip analysis copies of spans; canonical exported observations stay intact."""
+    if not until:
+        return events
+    end = timestamp(until).timestamp()
+    out = copy.deepcopy(events)
+    for event in out:
+        duration = max(0.0, float(event.get("duration_seconds") or 0))
+        if not duration:
+            continue
+        kept = min(duration, max(0.0, end - observed_timestamp(event["observed_at"]).timestamp()))
+        if kept < duration:
+            event["duration_seconds"] = kept
+            activity = (event.get("metadata") or {}).get("activity") or {}
+            for key in ("engaged_seconds", "idle_seconds", "active_input_seconds",
+                        "keypress_count", "click_count", "scroll_count"):
+                if key in activity:
+                    activity[key] *= kept / duration
+    return out
diff --git a/server/exporter.py b/server/exporter.py
index f0aab577..a7550248 100644
--- a/server/exporter.py
+++ b/server/exporter.py
@@ -12,7 +12,8 @@
 import xlsxwriter
 
 from .agent_export import AGENT_RUN_EXPORT_FIELDS, agent_run_export_rows
-from .analytics import candidate_tasks, semantic_activity, summary
+from .analytics import candidate_tasks, semantic_activity, summary, _summary_uncached
+from .export_window import selection, sql_bounds, analysis_rows
 from .db import normalized_rows, rows
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -43,6 +44,20 @@ def _read_bundled_text(path: Path, fallback: str) -> str:
 )
 
 
+def _analysis_request(window: dict[str, Any]) -> str:
+    dates = (f"{window['since']} to {window['until']} (end exclusive)" if window['scope'] == 'range'
+             else ('the current recording run' if window['scope'] == 'current' else 'all included retained evidence'))
+    return ("## Discover automation opportunities\n\n"
+            f"Analyze only {dates}. Find useful opportunities I have not identified upfront. "
+            "Inspect factual timelines and chronological events even when repeated-work candidates are empty. "
+            "For each opportunity give the observed work, proposed business outcome, supporting event IDs, "
+            "automatable portions, current capability checks and the next source-system investigation. "
+            "Use authorized source tools to resolve missing facts before asking targeted employee questions. "
+            "Separate observed effort from optional savings estimates and state every savings assumption. "
+            "Do not invent frequencies, business rules, completed outcomes or permission. "
+            "This file is evidence for analysis, not authorization to execute automation.")
+
+
 def _since_for_scope(scope: str) -> str | None:
     return os.getenv("WORKFLOW_OBSERVER_RUN_STARTED_AT") if scope == "current" else None
 
@@ -55,37 +70,23 @@ def _loader_for(table: str):
     raise ValueError("unsupported table")
 
 
-def _query_table(table: str, *, since: str | None) -> list[dict[str, Any]]:
-    """Return the complete evidence table for the requested scope.
-
-    Exports must never silently truncate evidence. Derived summaries may use bounded
-    analytical windows for responsiveness, but the exported evidence tables are
-    complete and the manifest states that explicitly.
-    """
+def _query_table(table: str, *, since: str | None, until: str | None = None) -> list[dict[str, Any]]:
     loader = _loader_for(table)
-    if since:
-        return loader(
-            f"SELECT * FROM {table} WHERE observed_at >= ? ORDER BY observed_at ASC",
-            (since,),
-        )
-    return loader(f"SELECT * FROM {table} ORDER BY observed_at ASC")
+    where, params = sql_bounds(since, until)
+    return loader(f"SELECT * FROM {table}" + (f" WHERE {where}" if where else "")
+                  + " ORDER BY julianday(observed_at), id", params)
 
 
-def _query_agent_events(*, since: str | None) -> list[dict[str, Any]]:
-    if since:
-        return rows(
-            "SELECT * FROM events WHERE source = ? AND observed_at >= ? ORDER BY observed_at ASC",
-            ("agent", since),
-        )
-    return rows("SELECT * FROM events WHERE source = ? ORDER BY observed_at ASC", ("agent",))
+def _query_agent_events(*, since: str | None, until: str | None = None) -> list[dict[str, Any]]:
+    where, params = sql_bounds(since, until)
+    return rows("SELECT * FROM events WHERE source = ?" + (f" AND {where}" if where else "")
+                + " ORDER BY julianday(observed_at), id", ("agent", *params))
 
 
-def _count_table(table: str, *, since: str | None) -> int:
+def _count_table(table: str, *, since: str | None, until: str | None = None) -> int:
     loader = _loader_for(table)
-    if since:
-        result = loader(f"SELECT COUNT(*) AS count FROM {table} WHERE observed_at >= ?", (since,))
-    else:
-        result = loader(f"SELECT COUNT(*) AS count FROM {table}")
+    where, params = sql_bounds(since, until)
+    result = loader(f"SELECT COUNT(*) AS count FROM {table}" + (f" WHERE {where}" if where else ""), params)
     return int((result[0] if result else {}).get("count", 0) or 0)
 
 
@@ -147,17 +148,26 @@ def _capture_manifest(
     }
 
 
-def build_export_payload(*, scope: str = "current", include_raw: bool = False) -> dict[str, Any]:
-    since = _since_for_scope(scope)
-    operational_summary = summary(100000, since=since, operational=True)
-    raw_summary = summary(100000, since=since, operational=False)
-    task_data = candidate_tasks(limit=100000, since=since)
-    operational_events = _query_table("normalized_events", since=since)
-    agent_events = _query_agent_events(since=since)
-    agent_runs = agent_run_export_rows(agent_events)
-    raw_events = _query_table("events", since=since) if include_raw else []
+def build_export_payload(*, scope: str = "current", include_raw: bool = False, since: str | None = None, until: str | None = None) -> dict[str, Any]:
+    window = selection(scope, since, until)
+    since, until = window["since"], window["until"]
+    operational_events = _query_table("normalized_events", since=since, until=until)
+    canonical = _query_table("events", since=since, until=until) if scope == "range" or include_raw else []
+    if scope == "range":
+        raw_analysis = analysis_rows(canonical, until)
+        op_analysis = analysis_rows(operational_events, until)
+        operational_summary = _summary_uncached(100000, operational=True, _raw_events=op_analysis)
+        raw_summary = _summary_uncached(100000, _raw_events=raw_analysis)
+        task_data = candidate_tasks(_raw_events=raw_analysis)
+    else:
+        operational_summary = summary(100000, since=since, operational=True)
+        raw_summary = summary(100000, since=since, operational=False)
+        task_data = candidate_tasks(limit=100000, since=since)
+    agent_events = _query_agent_events(since=since, until=until)
+    agent_runs = agent_run_export_rows(analysis_rows(agent_events, until))
+    raw_events = canonical if include_raw else []
     operational_count = len(operational_events)
-    raw_count = len(raw_events) if include_raw else _count_table("events", since=since)
+    raw_count = len(canonical) if scope == "range" or include_raw else _count_table("events", since=since)
     manifest = _capture_manifest(
         scope=scope,
         include_raw=include_raw,
@@ -167,6 +177,11 @@ def build_export_payload(*, scope: str = "current", include_raw: bool = False) -
         agent_run_rows=len(agent_runs),
     )
 
+    manifest["window"] = window | {"first_observed_at": raw_summary.get("first_observed_at"),
+                                   "last_observed_at": raw_summary.get("last_observed_at"),
+                                   "canonical_rows": raw_count,
+                                   "analysis_spans_clipped_to_end": bool(until),
+                                   "canonical_span_durations_preserved": True}
     return {
         "export": {
             "product": "OpenWorkGraph / Workflow Observer",
@@ -185,7 +200,7 @@ def build_export_payload(*, scope: str = "current", include_raw: bool = False) -
         },
         "capture_manifest": manifest,
         "ai_guide_markdown": AI_DATA_DICTIONARY_MD,
-        "starter_prompt_markdown": AI_STARTER_PROMPT_MD,
+        "starter_prompt_markdown": AI_STARTER_PROMPT_MD + "\n\n" + _analysis_request(window),
         "overview": {
             "operational": operational_summary,
             "raw_capture_counts": {
@@ -209,7 +224,7 @@ def build_export_payload(*, scope: str = "current", include_raw: bool = False) -
         "task_inference": task_data.get("inference", {}),
         "agent_runs": agent_runs,
         "operational_events": operational_events,
-        "operational_semantic_activity": semantic_activity(limit=100000, since=since, operational=True),
+        "operational_semantic_activity": semantic_activity(limit=100000, since=since, operational=True, **({"_raw_events": op_analysis} if scope == "range" else {})),
         **({"raw_local_evidence": raw_events} if include_raw else {}),
     }
 
@@ -250,6 +265,7 @@ def _flatten_event(e: dict[str, Any]) -> dict[str, Any]:
         "target_role": target_role,
         "duration_seconds": e.get("duration_seconds", 0),
         "metadata_json": json.dumps(meta, ensure_ascii=False, separators=(",", ":")),
+        "event_id": e.get("event_id"),
     }
 
 
diff --git a/server/local_auth.py b/server/local_auth.py
index 3cce38b9..0d196138 100644
--- a/server/local_auth.py
+++ b/server/local_auth.py
@@ -13,7 +13,7 @@
 _LOCK = threading.RLock()
 _DASHBOARD_SESSIONS: dict[str, None] = {}
 _CONSUMED_DASHBOARD_BOOTSTRAPS: set[str] = set()
-_EXPORT_TICKETS: dict[str, tuple[float, str, str, bool, bool]] = {}
+_EXPORT_TICKETS: dict[str, tuple[float, str, str, bool, bool, str, str]] = {}
 _PAIRING_CODE: tuple[str, float, int] | None = None
 _BROWSER_PAIR_REQUESTS: dict[str, tuple[float, str, bool]] = {}
 _SEEN_BROWSER_NONCES: dict[str, float] = {}
@@ -192,6 +192,7 @@ def issue_export_ticket(
     include_raw: bool,
     ttl_seconds: int = 30,
     redact_names: bool = False,
+    *, since: str | None = None, until: str | None = None,
 ) -> dict[str, int | str]:
     token = secrets.token_urlsafe(32)
     expires = time.time() + ttl_seconds
@@ -200,7 +201,7 @@ def issue_export_ticket(
         for key, value in list(_EXPORT_TICKETS.items()):
             if value[0] < now:
                 _EXPORT_TICKETS.pop(key, None)
-        _EXPORT_TICKETS[token] = (expires, str(export_format), str(scope), bool(include_raw), bool(redact_names))
+        _EXPORT_TICKETS[token] = (expires, str(export_format), str(scope), bool(include_raw), bool(redact_names), since or "", until or "")
     return {"ticket": token, "expires_in_seconds": ttl_seconds}
 
 
@@ -210,6 +211,7 @@ def consume_export_ticket(
     scope: str,
     include_raw: bool,
     redact_names: bool = False,
+    *, since: str | None = None, until: str | None = None,
 ) -> bool:
     if not token:
         return False
@@ -217,7 +219,7 @@ def consume_export_ticket(
         current = _EXPORT_TICKETS.pop(str(token), None)
         if not current:
             return False
-        expiry, expected_format, expected_scope, expected_raw, expected_redact = current
+        expiry, expected_format, expected_scope, expected_raw, expected_redact, expected_since, expected_until = current
         if expiry < time.time():
             return False
         return (
@@ -225,6 +227,8 @@ def consume_export_ticket(
             and hmac.compare_digest(str(scope), expected_scope)
             and bool(include_raw) is expected_raw
             and bool(redact_names) is expected_redact
+            and hmac.compare_digest((since or "").encode("utf-8"), expected_since.encode("utf-8"))
+            and hmac.compare_digest((until or "").encode("utf-8"), expected_until.encode("utf-8"))
         )
 
 
diff --git a/server/main.py b/server/main.py
index 5ad215aa..95598baa 100644
--- a/server/main.py
+++ b/server/main.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 import math
 import os
@@ -76,12 +77,27 @@ async def lifespan(_: FastAPI):
     # Idempotent local migration: remove legacy URL secrets and retroactively
     # apply browser exclusions before any API response can expose old rows.
     harden_existing_browser_events(_runtime_config())
-    from .db import protect_existing_titles_in_background
-
-    # One-time v0.108 title protection of older rows; batched and checkpointed,
-    # off the startup path so a long history never delays launch.
-    protect_existing_titles_in_background()
-    yield
+    from .db import (
+        mark_title_protection_complete_if_empty,
+        protect_existing_titles_in_background,
+    )
+    # A brand-new database needs no migration and can keep the original Full
+    # AI-context behavior immediately, without delaying startup.
+    mark_title_protection_complete_if_empty()
+
+    # Upgrades with 300k+ rows must not compete with the launcher's first
+    # /health probe (60s deadline). Schedule the strictly background worker
+    # only after ASGI startup can finish and the service can respond; it uses
+    # short, checkpointed transactions without blocking the event loop.
+    # During the delay and migration, Full AI context / raw export still
+    # downgrade to contextual redaction until the migration marker is set.
+    migration_timer = asyncio.get_running_loop().call_later(
+        3.0, protect_existing_titles_in_background,
+    )
+    try:
+        yield
+    finally:
+        migration_timer.cancel()
 
 
 app = FastAPI(title="OpenWorkGraph / Workflow Observer API", version=VERSION, lifespan=lifespan)
@@ -614,12 +630,16 @@ def get_session(session_id: str, limit: int = 1000):
 
 
 @app.get("/v1/export/{fmt}")
-def export_session(fmt: str, scope: str = "current", include_raw: bool = False, redact_names: bool = False):
+def export_session(fmt: str, scope: str = "current", include_raw: bool = False, redact_names: bool = False, since: str | None = None, until: str | None = None):
     fmt = fmt.lower().strip()
     if fmt not in {"json", "xlsx", "csvzip"}:
         raise HTTPException(status_code=400, detail="format must be json, xlsx, or csvzip")
-    if scope not in {"current", "all"}:
-        raise HTTPException(status_code=400, detail="scope must be current or all")
+    from .export_window import selection
+    try:
+        window = selection(scope, since, until)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from None
+    bounds = {"since": window["since"], "until": window["until"]} if scope == "range" else {}
     from .db import title_protection_complete
 
     if not title_protection_complete():
@@ -627,9 +647,9 @@ def export_session(fmt: str, scope: str = "current", include_raw: bool = False,
     if redact_names:
         # Same redactor as Redacted AI context: names/identifiers replaced in place.
         from .ai_context import redact_contextually
-        payload = redact_contextually(build_export_payload(scope=scope, include_raw=include_raw))
+        payload = redact_contextually(build_export_payload(scope=scope, include_raw=include_raw, **bounds))
     else:
-        payload = redact_for_display(build_export_payload(scope=scope, include_raw=include_raw))
+        payload = redact_for_display(build_export_payload(scope=scope, include_raw=include_raw, **bounds))
     if fmt == "json":
         body = json_bytes(payload)
         media = "application/json"
@@ -643,7 +663,7 @@ def export_session(fmt: str, scope: str = "current", include_raw: bool = False,
     return Response(
         content=body,
         media_type=media,
-        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+        headers={"Content-Disposition": f'attachment; filename="{filename}"', "Cache-Control": "no-store"},
     )
 
 
diff --git a/server/secure_app.py b/server/secure_app.py
index f00345bb..3564c62e 100644
--- a/server/secure_app.py
+++ b/server/secure_app.py
@@ -28,6 +28,8 @@
     verify_browser_authorization,
 )
 from .main import DASHBOARD, app
+from .export_window import selection
+from urllib.parse import urlencode
 
 COLLECTOR_ROUTES = {("POST", "/v1/events"), ("POST", "/v1/heartbeat")}
 BROWSER_ROUTES = {
@@ -177,7 +179,7 @@ def _connection_override_script() -> str:
       const d=await jsonCall('/v1/export-ticket',{
         method:'POST',
         headers:{'Content-Type':'application/json'},
-        body:JSON.stringify({format:String(fmt||''),scope:'current',include_raw:raw,redact_names:redactNames})
+        body:JSON.stringify({format:String(fmt||''),...(window.owgEvidenceSelection?.()||{scope:'current'}),include_raw:raw,redact_names:redactNames})
       });
       const a=document.createElement('a');
       a.href=d.url;
@@ -484,15 +486,20 @@ async def local_capability_guard(request: Request, call_next):
         scope = str(payload.get("scope") or "current").lower()
         include_raw = bool(payload.get("include_raw", True))
         redact_names = bool(payload.get("redact_names", False))
-        if export_format not in {"json", "xlsx", "csvzip"} or scope not in {"current", "all"}:
+        if export_format not in {"json", "xlsx", "csvzip"}:
             return _json_error("invalid export format or scope", 400)
-        issued = issue_export_ticket(export_format, scope, include_raw, redact_names=redact_names)
+        try:
+            window = selection(scope, payload.get("since"), payload.get("until"))
+        except ValueError as exc:
+            return _json_error(str(exc), 400)
+        bounds = {"since": window["since"], "until": window["until"]} if scope == "range" else {}
+        issued = issue_export_ticket(export_format, scope, include_raw, redact_names=redact_names, **bounds)
         raw = "true" if include_raw else "false"
         redact = "true" if redact_names else "false"
         ticket = str(issued["ticket"])
         return JSONResponse({
             **issued,
-            "url": f"/v1/export/{export_format}?scope={scope}&include_raw={raw}&redact_names={redact}&ticket={ticket}",
+            "url": f"/v1/export/{export_format}?" + urlencode({"scope": scope, "include_raw": raw, "redact_names": redact, "ticket": ticket, **bounds}),
         })
 
     if method == "GET" and path.startswith("/v1/export/") and request.query_params.get("ticket"):
@@ -501,7 +508,8 @@ async def local_capability_guard(request: Request, call_next):
         include_raw = str(request.query_params.get("include_raw") or "true").lower() == "true"
         redact_names = str(request.query_params.get("redact_names") or "false").lower() == "true"
         if not consume_export_ticket(
-            str(request.query_params.get("ticket") or ""), export_format, scope, include_raw, redact_names
+            str(request.query_params.get("ticket") or ""), export_format, scope, include_raw, redact_names,
+            since=request.query_params.get("since"), until=request.query_params.get("until"),
         ):
             return _json_error("invalid or expired export ticket", 401)
         return await call_next(request)
diff --git a/server/work_profile/__init__.py b/server/work_profile/__init__.py
index e15566ac..a283ab47 100644
--- a/server/work_profile/__init__.py
+++ b/server/work_profile/__init__.py
@@ -147,7 +147,7 @@ def list_self_tags(*, since: str | None = None, limit: int = 500) -> list[dict[s
     clauses: list[str] = []
     params: list[Any] = []
     if since:
-        clauses.append("ended_at >= ?")
+        clauses.append("julianday(ended_at) >= julianday(?)")
         params.append(since)
     where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
     with connect() as conn:
@@ -158,12 +158,15 @@ def list_self_tags(*, since: str | None = None, limit: int = 500) -> list[dict[s
     return [dict(row) | {"source": "voluntary_user_annotation"} for row in rows]
 
 
-def _focus_rows(since: str | None) -> list[dict[str, Any]]:
+def _focus_rows(since: str | None, *, until: str | None = None) -> list[dict[str, Any]]:
     params: tuple[Any, ...] = ()
     where = "WHERE n.event_type IN ('focus_span', 'focus_period')"
     if since:
-        where += " AND n.observed_at >= ?"
+        where += " AND julianday(n.observed_at) >= julianday(?)"
         params = (since,)
+    if until:
+        where += " AND julianday(n.observed_at) < julianday(?)"
+        params = (*params, until)
     with connect() as conn:
         rows = conn.execute(
             f"""
@@ -197,14 +200,17 @@ def _focus_rows(since: str | None) -> list[dict[str, Any]]:
     return result
 
 
-def _context_rows(since: str | None) -> list[dict[str, Any]]:
+def _context_rows(since: str | None, *, until: str | None = None) -> list[dict[str, Any]]:
     # The work profile describes the person's own work; agent runs are
     # reported separately (Agents tab, get_agent_runs).
     params: tuple[Any, ...] = ()
     where = "WHERE source != 'agent'"
     if since:
-        where += " AND observed_at >= ?"
+        where += " AND julianday(observed_at) >= julianday(?)"
         params = (since,)
+    if until:
+        where += " AND julianday(observed_at) < julianday(?)"
+        params = (*params, until)
     with connect() as conn:
         rows = conn.execute(
             f"""
@@ -566,13 +572,15 @@ def _hunting_candidates(context: list[dict[str, Any]]) -> list[dict[str, Any]]:
 def compute_work_profile(
     *, scope: str = "current", now: datetime | None = None,
     _preloaded_context: list[dict[str, Any]] | None = None,
+    _preloaded_focus: list[dict[str, Any]] | None = None,
+    _preloaded_tags: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     since = _scope_since(scope, now=now)
-    focus = _focus_rows(since)
+    focus = _focus_rows(since) if _preloaded_focus is None else _preloaded_focus
     context = _context_rows(since) if _preloaded_context is None else _preloaded_context
     transfers = _transfer_patterns(context)
     transfer_destinations = _transfer_destinations(transfers)
-    tags = list_self_tags(since=since)
+    tags = list_self_tags(since=since) if _preloaded_tags is None else _preloaded_tags
     profile = {
         "scope": scope,
         "since": since,
diff --git a/server/work_profile/service.py b/server/work_profile/service.py
index 89e5fe8e..c2ef9297 100644
--- a/server/work_profile/service.py
+++ b/server/work_profile/service.py
@@ -11,6 +11,7 @@
     list_self_tags,
     compute_work_profile as _compute_work_profile,
     _context_rows,
+    _focus_rows,
     _scope_since,
 )
 from .accuracy import compute_today_work_profile, local_day_since
@@ -51,8 +52,30 @@ def backfill_clipboard_context_links() -> int:
     return changed
 
 
-def compute_work_profile(*, scope: str = "current", now=None) -> dict[str, Any]:
+def compute_work_profile(*, scope: str = "current", now=None, since: str | None = None, until: str | None = None) -> dict[str, Any]:
     backfill_clipboard_context_links()
+    if scope == "range":
+        from server.export_window import selection, timestamp, observed_timestamp
+        window = selection(scope, since, until)
+        since, until = window["since"], window["until"]
+        context = _context_rows(since, until=until)
+        focus = _focus_rows(since, until=until)
+        end = timestamp(until).timestamp()
+        for row in focus:
+            duration = row["foreground_seconds"]
+            kept = min(duration, max(0, end - observed_timestamp(row["observed_at"]).timestamp()))
+            if duration and kept < duration:
+                for key in ("engaged_seconds", "keypress_count", "click_count", "scroll_count"):
+                    row[key] *= kept / duration
+                row["duration_seconds"] = row["foreground_seconds"] = kept
+        tags = [t for t in list_self_tags(since=since) if observed_timestamp(t["started_at"]) < timestamp(until)]
+        for tag in tags:
+            tag["started_at"] = max(observed_timestamp(tag["started_at"]), timestamp(since)).isoformat()
+            tag["ended_at"] = min(observed_timestamp(tag["ended_at"]), timestamp(until)).isoformat()
+        profile = _compute_work_profile(scope="all", _preloaded_context=context,
+                                       _preloaded_focus=focus, _preloaded_tags=tags)
+        profile.update(scope="range", since=since, until=until)
+        return enrich_work_profile(profile, since=since, rows=context)
     since = local_day_since(now=now) if scope == "today" else _scope_since(scope, now=now)
     # Read and decode canonical context once, then feed the same material to
     # both core profile metrics and optional browser-signal enrichments.
diff --git a/shared/core/sensitive_identifiers.py b/shared/core/sensitive_identifiers.py
index 3bd5d248..48140072 100644
--- a/shared/core/sensitive_identifiers.py
+++ b/shared/core/sensitive_identifiers.py
@@ -541,6 +541,19 @@ def walk(value: Any, *, key: str = "") -> Any:
         if isinstance(value, tuple):
             return tuple(walk(v, key=key) for v in value)
         if isinstance(value, str):
+            # A validated opaque hash can coincidentally contain a Luhn-valid
+            # digit sequence. Redacting it breaks project/file linkage.
+            # Only this typed field and exact hash format receive protection.
+            if key == "workspace_ref":
+                from shared.tool_detail import valid_workspace_ref
+
+                if valid_workspace_ref(value):
+                    return value
+            if key in {"file_ref", "file_refs"}:
+                from shared.tool_detail import sanitize_detail
+
+                if sanitize_detail({"file_refs": [value]}).get("file_refs") == [value]:
+                    return value
             if structured_identifier_kind(key):
                 return tokenize_structured_identifier(key, value)
             return redact_sensitive_identifiers(value, redact_adjacent_name=False)
diff --git a/tests/js/analysis_period.test.mjs b/tests/js/analysis_period.test.mjs
new file mode 100644
index 00000000..ca36261f
--- /dev/null
+++ b/tests/js/analysis_period.test.mjs
@@ -0,0 +1,31 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import vm from 'node:vm';
+const source=fs.readFileSync(new URL('../../dashboard/analysis_handoff.js',import.meta.url),'utf8');
+function periodAt(instant,zone) {
+  const previous=process.env.TZ;process.env.TZ=zone;
+  try {
+    class FixedDate extends Date { constructor(...args) {super(...(args.length?args:[instant]));} }
+    const window={};
+    vm.runInNewContext(source,{window,document:{readyState:'loading',addEventListener(){}},Date:FixedDate,Intl});
+    return JSON.parse(JSON.stringify(window.owgEvidenceSelection()));
+  } finally {if(previous===undefined)delete process.env.TZ;else process.env.TZ=previous;}
+}
+test('last seven local calendar days span 169 hours across the New York DST change',()=>{
+  const p=periodAt('2026-11-01T12:00:00Z','America/New_York');
+  assert.equal(p.since,'2026-10-26T04:00:00.000Z');
+  assert.equal(p.until,'2026-11-02T05:00:00.000Z');
+  assert.equal((Date.parse(p.until)-Date.parse(p.since))/3600000,169);
+});
+test('fractional-offset timezones retain local calendar boundaries',()=>{
+  const p=periodAt('2026-10-16T12:00:00Z','Asia/Kathmandu');
+  assert.equal(p.since,'2026-10-09T18:15:00.000Z');
+  assert.equal(p.until,'2026-10-16T18:15:00.000Z');
+});
+test('UTC period includes today and six previous days without requesting all history',()=>{
+  const p=periodAt('2026-10-16T12:00:00Z','UTC');
+  assert.equal(p.scope,'range');
+  assert.equal(p.since,'2026-10-10T00:00:00.000Z');
+  assert.equal(p.until,'2026-10-17T00:00:00.000Z');
+});
diff --git a/tests/test_agent_brief_v0103.py b/tests/test_agent_brief_v0103.py
index a13437da..234ace4d 100644
--- a/tests/test_agent_brief_v0103.py
+++ b/tests/test_agent_brief_v0103.py
@@ -234,3 +234,32 @@ def boom(_payload):
 
     assert run_main({"source": "startup"}, boom) == ""
     assert run_main({"source": "startup"}, lambda p: "") == ""
+
+
+def test_card_like_workspace_hash_preserves_project_context(tmp_path):
+    _run(r'''
+from shared.core.sensitive_identifiers import sanitize_event_identifiers
+import shared.tool_detail as detail
+original_workspace_ref = detail.workspace_ref
+# Valid opaque hash containing a Luhn-valid card number: reproduce the rare
+# collision deterministically without relying on this machine's random key.
+ref = "w:4111111111111111"
+detail.workspace_ref = lambda path: ref if path == PROJECT_A else original_workspace_ref(path)
+run("a1", PROJECT_A, 0, "8 passed in 1s")
+run("b1", PROJECT_B, 10, "1 failed in 1s", command="npm test")
+brief = agent_brief.build_brief("claude-code", workspace_ref=ref)
+assert brief["scope"] == "this_project" and brief["runs_considered"] == 1, brief
+assert "npm" not in brief["text"]
+safe = sanitize_event_identifiers({"window_title": "4111111111111111", "metadata": {
+    "workspace_ref": ref, "other": "4111111111111111",
+    "tool_detail": {"file_refs": ["f:4111111111111111", "4111111111111111"]},
+    "file_ref": "f:4111111111111111",
+    "nested": {"workspace_ref": "4111111111111111"}}})
+assert safe["metadata"]["workspace_ref"] == ref
+assert safe["metadata"]["file_ref"] == "f:4111111111111111"
+assert safe["metadata"]["tool_detail"]["file_refs"][0] == "f:4111111111111111"
+assert "4111111111111111" not in safe["metadata"]["tool_detail"]["file_refs"][1]
+assert "4111111111111111" not in safe["window_title"]
+assert "4111111111111111" not in safe["metadata"]["other"]
+assert "4111111111111111" not in safe["metadata"]["nested"]["workspace_ref"]
+''', tmp_path)
diff --git a/tests/test_dashboard_hardening_followup.py b/tests/test_dashboard_hardening_followup.py
index 563b9f46..785bba1f 100644
--- a/tests/test_dashboard_hardening_followup.py
+++ b/tests/test_dashboard_hardening_followup.py
@@ -105,7 +105,7 @@ def test_empty_base_csv_exports_keep_real_header_rows_without_fake_data():
             "transitions.csv": "from,to,count",
             "inferred_tasks.csv": "interpretation_status,started_at,ended_at,task_label,task_family,confidence,primary_surface,surfaces,elapsed_seconds,foreground_seconds,engaged_seconds,idle_seconds,active_input_seconds,keypress_count,click_count,scroll_count,boundary_reason,boundary_confidence,effort_estimated,semantic_actions",
             "repeated_task_families.csv": "interpretation_status,task_label,task_family,observed_count,surfaces,surface_variant_count,total_engaged_seconds,median_engaged_seconds,p90_engaged_seconds,median_elapsed_seconds,keypress_count,click_count,completion_boundary_count,confidence",
-            "operational_events.csv": "observed_at,schema_version,organization_id,actor_id,device_id,sensor_id,source,session_id,app_or_surface,window_title,event_type,action,page_host,page_path,target_label,target_role,duration_seconds,metadata_json",
+            "operational_events.csv": "observed_at,schema_version,organization_id,actor_id,device_id,sensor_id,source,session_id,app_or_surface,window_title,event_type,action,page_host,page_path,target_label,target_role,duration_seconds,metadata_json,event_id",
             "semantic_activity.csv": "observed_at,session_id,app,window_title,event_type,label,page,target,source",
         }
         for filename, header in expected.items():
diff --git a/tests/test_dated_ai_handoff.py b/tests/test_dated_ai_handoff.py
new file mode 100644
index 00000000..dd7ab805
--- /dev/null
+++ b/tests/test_dated_ai_handoff.py
@@ -0,0 +1,280 @@
+"""Cross-layer regressions: selected evidence, privacy, and scoped download capabilities."""
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def run(code, tmp_path):
+    env = dict(os.environ, WORKFLOW_OBSERVER_DATA=str(tmp_path/'data'),
+               WORKFLOW_OBSERVER_AUTH_DIR=str(tmp_path/'auth'),
+               WORKFLOW_OBSERVER_RUN_STARTED_AT='2026-10-17T00:00:00+00:00')
+    p = subprocess.run([sys.executable, '-c', SETUP+'\n'+code], cwd=ROOT,
+                       env=env, capture_output=True, text=True, timeout=120)
+    assert p.returncode == 0, p.stdout+'\n'+p.stderr
+
+
+SETUP = r'''
+import io, json, zipfile
+from fastapi.testclient import TestClient
+import server.enterprise_app, server.history_routes, server.v0571_polish
+import server.evidence_paging, server.work_profile_routes, server.basic_mode_routes
+from server.secure_app import app
+from server.db import insert_events, connect
+from server.local_auth import ensure_api_token, create_dashboard_session
+from server.context_exporter import build_export_payload, csv_zip_bytes, xlsx_bytes
+H={'Authorization':'Bearer '+ensure_api_token()}
+D={'Authorization':'OWG-Session '+create_dashboard_session()}
+A='2026-10-16T09:00:00+00:00'
+B='2026-10-16T10:00:00+00:00'
+def event(eid, at, title, duration=30, **extra):
+    return dict(event_id=eid, observed_at=at,device_id='synthetic',session_id='s',
+                app='Microsoft Excel',window_title=title,event_type='focus_span',
+                duration_seconds=duration,metadata={'activity':{'engaged_seconds':duration,'click_count':4}},**extra)
+'''
+
+
+def test_dated_package_uses_same_window_for_all_layers_and_formats(tmp_path):
+    run(r'''
+with TestClient(app) as c:
+    insert_events([
+      event('before','2026-10-16T08:59:59Z','OUTSIDE_BEFORE_SENTINEL'),
+      event('inside','2026-10-16T10:00:00+01:00','Invoice Acme Q4',7200),
+      event('inside2','2026-10-16T09:30:00Z','Invoice Acme Q4'),
+      event('after',B,'OUTSIDE_AFTER_SENTINEL'),
+    ])
+    current=c.get('/v1/export/json?scope=current&include_raw=true',headers=H).json()
+    assert current['raw_local_evidence']==[]
+    args=dict(scope='range',since=A,until=B,include_raw=True,redact_names=True)
+    r=c.get('/v1/export/json',params=args,headers=H)
+    assert r.status_code==200,r.text
+    data=r.json()
+    assert {e['event_id'] for e in data['raw_local_evidence']}=={'inside','inside2'}
+    assert 'OUTSIDE_BEFORE_SENTINEL' not in r.text and 'OUTSIDE_AFTER_SENTINEL' not in r.text
+    assert data['raw_local_evidence'][0]['duration_seconds']==7200  # canonical record unchanged
+    assert data['overview']['operational']['total_active_seconds']==3630
+    assert data['work_profile']['fragmentation']['foreground_seconds']==3630
+    assert data['capture_manifest']['window']['until']==B
+    assert data['capture_manifest']['window']['canonical_rows']==2
+    assert data['capture_manifest']['window']['analysis_spans_clipped_to_end']
+    assert 'no repeated' not in data['starter_prompt_markdown'].lower() # no repetition gate
+    assert 'even when repeated-work candidates are empty' in data['starter_prompt_markdown']
+    for fmt in ('csvzip','xlsx'):
+        r=c.get('/v1/export/'+fmt,params=args,headers=H)
+        assert r.status_code==200,r.text
+        with zipfile.ZipFile(io.BytesIO(r.content)) as z:
+            content=b'\n'.join(z.read(n) for n in z.namelist())
+            assert b'OUTSIDE_AFTER_SENTINEL' not in content and b'OUTSIDE_BEFORE_SENTINEL' not in content
+            if fmt=='csvzip':
+                assert b'event_id' in z.read('raw_local_evidence.csv')
+                assert 'START_HERE.md' in z.namelist()
+                manifest=json.loads(z.read('CAPTURE_MANIFEST.json'))
+                assert manifest['window']['canonical_rows']==2
+    with connect() as conn:
+        assert conn.execute("SELECT duration_seconds FROM events WHERE event_id='inside'").fetchone()[0]==7200
+''', tmp_path)
+
+
+def test_download_ticket_cannot_widen_dates_and_remains_one_use(tmp_path):
+    run(r'''
+with TestClient(app) as c:
+    insert_events([event('inside',A,'Invoice Acme')])
+    args=dict(format='json',scope='range',since=A,until=B,include_raw=True,redact_names=True)
+    for replacement in ('scope=all','until=2026-10-17T00%3A00%3A00%2B00%3A00','redact_names=false','since=猫'):
+        r=c.post('/v1/export-ticket',headers=D,json=args)
+        assert r.status_code==200,r.text
+        url=r.json()['url']
+        from urllib.parse import urlsplit,parse_qs,urlencode
+        query=parse_qs(urlsplit(url).query)
+        key,val=replacement.split('=',1)
+        from urllib.parse import unquote
+        query[key]=[unquote(val)]
+        bad='/v1/export/json?'+urlencode({k:v[0] for k,v in query.items()})
+        assert c.get(bad).status_code==401
+        assert c.get(url).status_code==401
+    good=c.post('/v1/export-ticket',headers=D,json=args).json()['url']
+    assert c.get(good).status_code==200
+    assert c.get(good).status_code==401
+    for payload in [dict(args,until=A),dict(args,since='bad'),dict(args,since='2026-10-16T09:00:00')]:
+        assert c.post('/v1/export-ticket',headers=D,json=payload).status_code==400
+''', tmp_path)
+
+
+def test_preview_does_not_grant_history_or_depend_on_repeated_work(tmp_path):
+    run(r'''
+from shared.history_policy import active_ai_history_access
+with TestClient(app) as c:
+    insert_events([event('inside',A,'natalie bergstrom cardiology - Google Search')])
+    prior=active_ai_history_access()
+    args=dict(scope='range',since=A,until=B)
+    assert c.get('/v1/analysis-preview',params=args).status_code==401
+    assert c.get('/v1/analysis-preview',headers=H,params=args).status_code==403
+    r=c.get('/v1/analysis-preview',headers=D,params=args)
+    assert r.status_code==200,r.text
+    d=r.json()
+    assert d['canonical_rows']==1 and d['retained_rows']==1
+    assert 'natalie' not in r.text and 'bergstrom' not in r.text
+    assert 'cardiology' in r.text and 'PERSON_' in r.text
+    assert d['work_text']['included_in_activity_package'] is False
+    assert active_ai_history_access()==prior
+    empty=c.get('/v1/analysis-preview',headers=D,params={'scope':'current'}).json()
+    assert empty['canonical_rows']==0 and empty['retained_rows']==1
+    assert 'without a search filter' in d['request_text']
+''', tmp_path)
+
+
+def test_activity_paging_respects_dates_offsets_and_cursor_binding(tmp_path):
+    run(r'''
+with TestClient(app) as c:
+    insert_events([event('before','2026-10-16T08:59:59Z','OUTSIDE_BEFORE_SENTINEL'),
+                   event('one','2026-10-16T10:00:00+01:00','Invoice Acme'),
+                   event('two','2026-10-16T09:30:00Z','Receipt Acme'),
+                   event('after',B,'OUTSIDE_AFTER_SENTINEL')])
+    args=dict(scope='range',since=A,until=B,limit=1)
+    r=c.get('/v1/evidence',headers=H,params=args)
+    assert r.status_code==200,r.text
+    first=r.json();assert first['total']==2,first
+    assert first['items'][0]['event_id']=='two'
+    cursor=first['next_cursor'];assert cursor
+    next=c.get('/v1/evidence',headers=H,params=dict(args,cursor=cursor)).json()
+    assert next['items'][0]['event_id']=='one'
+    assert c.get('/v1/evidence',headers=H,params=dict(args,cursor=cursor,until='2026-10-17T00:00:00Z')).status_code==400
+    # Preserve the existing recorder-state boundary for legacy current-run calls,
+    # even when it differs from the launcher's process environment.
+    from unittest.mock import patch
+    with patch('server.enterprise_app._current_run_start', return_value=A):
+        current=c.get('/v1/evidence',headers=H,params={'scope':'current'}).json()
+    assert current['since']==A and current['total']==3, current
+    assert {x['event_id'] for x in current['items']}=={'one','two','after'}
+''', tmp_path)
+
+
+def test_lowercase_search_names_are_hidden_without_losing_context(tmp_path):
+    run(r'''
+# Do not depend on HTTP import-order to enable capture-time redaction.
+from shared.core.browser_title_privacy import protect_text
+from server.ai_context import redact_contextually, contextual_text_redactor
+for title in ('natalie bergstrom cardiology - Google Search',
+              'anna svensson acme - Sök på Google - Google Chrome'):
+    for transform in (protect_text, lambda x:redact_contextually({'window_title':x})['window_title']):
+        safe=transform(title)
+        assert 'PERSON_' in safe,(title,safe)
+        assert 'natalie' not in safe and 'bergstrom' not in safe and 'anna' not in safe and 'svensson' not in safe
+        assert 'cardiology' in safe or 'acme' in safe
+        assert transform(safe)==safe
+for title in ('Q4 pipeline - Atlas Logistics - Google Sheets',
+              'Fakturamall Sollefteå v42 Acme AB - Google Sheets',
+              'mark as read - Google Search','Stockholm weather - Google Search','De Beers','Van rental'):
+    assert protect_text(title)==title,(title,protect_text(title))
+r=contextual_text_redactor({'never_redact':['anna svensson'],'always_redact':[]})
+assert r('anna svensson cardiology - Google Search')=='anna svensson cardiology - Google Search'
+''', tmp_path)
+
+
+def test_upgrade_reprotects_history_after_the_previous_title_migration(tmp_path):
+    run(r'''
+from server import db
+db.init_db()
+insert_events([event('legacy', A, 'Invoice Acme Q4')])
+with connect() as c:
+    # Reproduce a retained title missed by the already-completed older pass.
+    c.execute("UPDATE events SET window_title=? WHERE event_id='legacy'",
+              ('natalie bergstrom cardiology - Google Search',))
+    c.execute("INSERT OR IGNORE INTO privacy_migrations(migration_key) VALUES (?)",
+              ('titles_sensitive_details_tokenized_v0108',))
+assert not db.title_protection_complete()
+assert db.protect_existing_titles(batch_size=1) == 1
+assert db.title_protection_complete()
+with connect() as c:
+    for table, column in [('events','window_title'), ('normalized_events','window_title'),
+                          ('context_events','resource_title')]:
+        title=c.execute(f"SELECT {column} FROM {table} WHERE event_id='legacy'").fetchone()[0]
+        assert 'natalie' not in title and 'bergstrom' not in title, (table,title)
+        if table != 'normalized_events':  # normalized titles intentionally use coarse app labels
+            assert 'PERSON_' in title and 'cardiology' in title, (table,title)
+assert db.protect_existing_titles(batch_size=1) == 0
+''', tmp_path)
+
+
+def test_upgrade_responds_to_health_before_heavy_title_migration(tmp_path):
+    run(r'''
+import time
+from threading import Event
+from unittest.mock import patch
+from server import db
+started=Event()
+finished=Event()
+# Simulate an existing installation with a pending historical re-protection.
+db.init_db()
+insert_events([event('prior_history', A, 'Invoice for Acme')])
+with connect() as c:
+    c.execute("UPDATE events SET window_title=? WHERE event_id='prior_history'",
+              ('natalie bergstrom cardiology - Google Search',))
+assert db.title_protection_complete() is False
+def expensive_history_scan(*,batch_size,pause_seconds):
+    # Models a 300k-event migration without spending CI minutes writing fixtures.
+    assert batch_size <= 200
+    assert pause_seconds >= 0.01
+    started.set()
+    time.sleep(1.0)
+    finished.set()
+with patch.object(db,'protect_existing_titles',expensive_history_scan):
+    began=time.monotonic()
+    with TestClient(app) as c:
+        # ASGI lifespan finishes before migration. The launcher sees /health.
+        assert time.monotonic()-began < 2.5
+        response=c.get('/health')
+        assert response.status_code == 200,response.text
+        assert not started.is_set(), 'title migration launched before health became available'
+        assert db.title_protection_complete() is False
+        assert started.wait(5), 'deferred migration never started'
+        check=time.monotonic()
+        assert c.get('/health').status_code == 200
+        assert time.monotonic()-check < 0.8, 'slow migration blocked health'
+        assert finished.wait(5)
+        assert db.title_protection_complete() is False
+''', tmp_path)
+
+
+def test_title_migration_checkpoint_yields_and_preserves_privacy_marker(tmp_path):
+    run(r'''
+from server import db
+from unittest.mock import patch
+db.init_db()
+insert_events([
+  event('old1', A, 'Invoice for Acme'),
+  event('old2', A, 'Invoice for Acme'),
+  event('old3', A, 'Invoice for Acme'),
+])
+with connect() as c:
+    for eid in ('old1','old2','old3'):
+        c.execute('UPDATE events SET window_title=? WHERE event_id=?',
+                  ('natalie bergstrom cardiology - Google Search', eid))
+assert db.title_protection_complete() is False
+pauses=[]
+with patch.object(db.time,'sleep',lambda seconds: pauses.append(seconds)):
+    assert db.protect_existing_titles(batch_size=1,pause_seconds=0.02) == 3
+assert len(pauses)==3 and all(0 < value <= 0.2 for value in pauses), pauses
+assert db.title_protection_complete() is True
+with connect() as c:
+    assert c.execute('SELECT COUNT(*) FROM events').fetchone()[0] == 3
+    for table,column in [('events','window_title'),('normalized_events','window_title'),('context_events','resource_title')]:
+        result=[row[0] for row in c.execute(f'SELECT {column} FROM {table}')]
+        assert len(result)==3,(table,result)
+        assert not any('natalie' in value or 'bergstrom' in value for value in result)
+''', tmp_path)
+
+
+
+def test_empty_install_title_marker_does_not_delay_full_context(tmp_path):
+    run(r'''
+from server import db
+assert db.title_protection_complete() is False
+with TestClient(app) as c:
+    assert c.get('/health').status_code == 200
+    assert db.title_protection_complete() is True
+    with connect() as conn:
+        assert conn.execute('SELECT COUNT(*) FROM events').fetchone()[0] == 0
+''', tmp_path)
diff --git a/tests/test_migration_window_v0108.py b/tests/test_migration_window_v0108.py
index 98a682bf..68115f41 100644
--- a/tests/test_migration_window_v0108.py
+++ b/tests/test_migration_window_v0108.py
@@ -31,12 +31,15 @@ def test_full_context_and_unredacted_exports_wait_for_the_title_migration(tmp_pa
 H = {"Authorization": f"Bearer {ensure_api_token()}"}
 AI = {**H, "X-OpenWorkGraph-Context": "ai"}
 now = datetime.now(timezone.utc).isoformat()
+# A real upgrade already contains the unprotected v0.107 row *before* startup.
+# An empty fresh installation legitimately has its privacy marker completed
+# immediately, so injecting a "legacy" row after first launch is not realistic.
+db.init_db()
+with db.connect() as conn:
+    conn.execute("INSERT INTO events(event_id, observed_at, device_id, session_id, app, window_title, event_type, duration_seconds, metadata_json)"
+                 " VALUES ('old', ?, 'd', 's', 'Microsoft Teams', 'Chat with Anna Svensson about Q4 renewal', 'focus_span', 30, '{}')", (now,))
 
 with TestClient(app) as c:
-    # 1. A row stored by v0.107, before titles were protected.
-    with db.connect() as conn:
-        conn.execute("INSERT INTO events(event_id, observed_at, device_id, session_id, app, window_title, event_type, duration_seconds, metadata_json)"
-                     " VALUES ('old', ?, 'd', 's', 'Microsoft Teams', 'Chat with Anna Svensson about Q4 renewal', 'focus_span', 30, '{}')", (now,))
     assert not db.title_protection_complete()
 
     # 2. The person had chosen Full AI context.
@@ -69,22 +72,29 @@ def test_full_context_and_unredacted_exports_wait_for_the_title_migration(tmp_pa
     assert result.returncode == 0, f"stdout={result.stdout}\nstderr={result.stderr}"
 
 
-def test_small_histories_are_protected_before_startup_returns(tmp_path):
+def test_small_histories_migrate_in_background_without_blocking(tmp_path):
     env = os.environ.copy()
     env.update({"WORKFLOW_OBSERVER_DATA": str(tmp_path / "data"), "WORKFLOW_OBSERVER_AUTH_DIR": str(tmp_path / "auth"),
                 "WORKFLOW_OBSERVER_CONFIG": str(tmp_path / "config.json")})
     code = r'''
 from datetime import datetime, timezone
+import time
 import server.db as db
 db.init_db()
 with db.connect() as conn:
     conn.execute("INSERT INTO events(event_id, observed_at, device_id, session_id, app, window_title, event_type, duration_seconds, metadata_json)"
                  " VALUES ('old', ?, 'd', 's', 'Slack', 'Erik Lund (DM) - Slack', 'focus_span', 30, '{}')", (datetime.now(timezone.utc).isoformat(),))
+assert not db.title_protection_complete()
+began=time.monotonic()
 db.protect_existing_titles_in_background()
-assert db.title_protection_complete()  # synchronous: nothing left for a background thread
+assert time.monotonic()-began < 0.8, "a small legacy history also must not block the caller"
+deadline=time.monotonic()+15
+while not db.title_protection_complete() and time.monotonic() < deadline:
+    time.sleep(0.04)
+assert db.title_protection_complete(), "short history did not eventually complete"
 with db.connect() as conn:
     assert "Erik" not in conn.execute("SELECT window_title FROM events").fetchone()[0]
-db.protect_existing_titles_in_background()  # already done: returns at once
+db.protect_existing_titles_in_background()  # completed migration remains idempotent
 '''
     result = subprocess.run([sys.executable, "-c", code], cwd=ROOT, env=env, text=True, capture_output=True, timeout=180)
-    assert result.returncode == 0, f"stdout={result.stdout}\nstderr={result.stderr}"
+    assert result.returncode == 0, f"stdout={result.stdout}\\nstderr={result.stderr}"
diff --git a/tests/test_v47_dashboard_mcp.py b/tests/test_v47_dashboard_mcp.py
index ab63e2bf..a467ede2 100644
--- a/tests/test_v47_dashboard_mcp.py
+++ b/tests/test_v47_dashboard_mcp.py
@@ -95,7 +95,7 @@ def test_dashboard_data_first_layout_keeps_required_observer_controls():
     for panel in ("overview", "evidence", "connect", "organization", "export"):
         assert f'data-panel="{panel}"' in html
     assert "Reset learned person aliases" in html
-    assert "Export this run" in html
+    assert "Export selected work" in html
     assert 'id="includeRawTop"' not in html
     assert "Connect AI" in html
     assert "Add to Cursor" in html