diff --git a/plugins/session-flow/.claude-plugin/plugin.json b/plugins/session-flow/.claude-plugin/plugin.json
index 6de0e5b900..28e8e9d4f3 100644
--- a/plugins/session-flow/.claude-plugin/plugin.json
+++ b/plugins/session-flow/.claude-plugin/plugin.json
@@ -1,6 +1,6 @@
 {
   "name": "session-flow",
-  "version": "0.48.5",
+  "version": "0.48.6",
   "description": "Session lifecycle: workflow (next stage), handoff (save-point, resume prompt), continue-in-background, keep-going (resume after interruption or stall), find-handoff (recover a lost handoff), clean-stop (durable stopping point), retro and running-retro (retrospectives), audit-sessions, orient (where the session stands), orchestrate (delegation imperatives), reanchor (verify assumptions), reconcile (retire finished work), show-options (ranked skill menu), tidy-work (.work tiers), check, setup.",
   "author": {
     "name": "Melodic Software",
diff --git a/plugins/session-flow/CHANGELOG.md b/plugins/session-flow/CHANGELOG.md
index 1de9ee9fce..6edb368457 100644
--- a/plugins/session-flow/CHANGELOG.md
+++ b/plugins/session-flow/CHANGELOG.md
@@ -1,5 +1,15 @@
 # Changelog: session-flow plugin
 
+## [0.48.6] - 2026-10-04
+
+### Fixed
+
+- **`/session-flow:audit-sessions` takes its medians and findings over interactive sessions only ([#6076](https://github.com/melodic-software/claude-code-plugins/issues/6076)).**
+  Each store record now carries the transcript's `entrypoint` values in `entrypoints`, or `["unknown"]`. Sweep sets aside a session whose entrypoints are all headless or Agent SDK ones (`sdk-cli`, `sdk-ts`, `sdk-py`), says how many in the report, and keeps a session with no known entrypoint, counted as unclassified; drift still reads every record. A new canary, `key_path:user:entrypoint`, withholds the four metrics whose medians automated sessions were measured to move if Claude Code stops writing the key.
+- **One interrupt in a session is no longer a `tools.interrupts` finding.** The threshold is now 1, so a session needs two.
+- **Claude Desktop prompts count as typed turns.** Desktop writes the person's prompts with `promptSource: sdk`, which the shared transcript reader rejected. A record whose `origin.kind` is `human` now counts whatever its `promptSource`, after one leading `<system-reminder>` block is dropped; `/session-flow:retro` counts them too. CLI records count as before.
+- **A transcript Claude Code set aside as `<session>.orphaned-*.jsonl` is no longer collected as a second session.** Collect skips it and reports the count in `skipped_orphaned`, and deletes a record an earlier version stored for one, even when the transcript is gone, reporting the count in `purged_orphaned`.
+
 ## [0.48.5] - 2026-10-04
 
 ### Fixed
diff --git a/plugins/session-flow/scripts/transcript_reader.py b/plugins/session-flow/scripts/transcript_reader.py
index e9c3d82237..dde0e8e7fd 100644
--- a/plugins/session-flow/scripts/transcript_reader.py
+++ b/plugins/session-flow/scripts/transcript_reader.py
@@ -15,10 +15,13 @@
   plus `unique_messages`. Non-integer counts read as 0.
 - `user_text(record)` is a user record's text, stripped, with text blocks joined by newlines; None
   for a tool result or a content shape that carries no text. Injected text is returned as is.
-- `is_typed_turn(record)` is true only for a user record the human typed: not a tool result, not
-  meta, compact-summary or transcript-only, `origin.kind` absent or `human`, `promptSource` absent
-  or in `TYPED_PROMPT_SOURCES`, not an interrupt, and not starting with an `INJECTED_PREFIXES`
-  entry. Injected records outnumber typed turns.
+- `typed_text(record)` is the text of a user record the human typed, else None: not a tool
+  result, not meta, compact-summary or transcript-only, `origin.kind` absent or `human`,
+  `promptSource` absent or in `TYPED_PROMPT_SOURCES` unless `origin.kind` is `human`, not an
+  interrupt, and not starting with an `INJECTED_PREFIXES` entry. For an `origin.kind: human` record,
+  one leading `<system-reminder>` block is dropped first: Claude Desktop writes the person's prompts
+  that way, with `promptSource: sdk`. Injected records outnumber typed turns.
+- `is_typed_turn(record)` is `typed_text(record) is not None`.
 - `iter_subagents(main_path)` yields a `Subagent(path, meta)` for each
   `<session>/subagents/agent-*.jsonl` beside `<session>.jsonl`, in name order; `meta` is the
   parsed `agent-*.meta.json` object, or None when it is missing, unreadable or not an object.
@@ -67,6 +70,7 @@
     "<user-prompt-submit-hook>",
 )
 INTERRUPT_RE = re.compile(r"^\[Request interrupted by user( for tool use)?\]")
+LEADING_REMINDER_RE = re.compile(r"<system-reminder>.*?</system-reminder>\s*", re.S)
 TYPED_PROMPT_SOURCES = {"typed", "queued"}
 RECORD_TYPES = frozenset(
     {
@@ -177,18 +181,24 @@ def user_text(record: dict) -> str | None:
     return content.strip() if isinstance(content, str) else None
 
 
-def is_typed_turn(record: dict) -> bool:
+def typed_text(record: dict) -> str | None:
     if record.get("type") != "user":
-        return False
+        return None
     if record.get("isMeta") or record.get("isCompactSummary") or record.get("isVisibleInTranscriptOnly"):
-        return False
+        return None
     kind = _obj(record.get("origin")).get("kind")
     if kind and kind != "human":
-        return False
+        return None
     source = record.get("promptSource")
-    if source and not (isinstance(source, str) and source in TYPED_PROMPT_SOURCES):
-        return False
+    if kind != "human" and source and not (isinstance(source, str) and source in TYPED_PROMPT_SOURCES):
+        return None
     text = user_text(record)
-    if not text or INTERRUPT_RE.match(text):
-        return False
-    return not text.startswith(INJECTED_PREFIXES)
+    if text and kind == "human" and (reminder := LEADING_REMINDER_RE.match(text)):
+        text = text[reminder.end() :]
+    if not text or INTERRUPT_RE.match(text) or text.startswith(INJECTED_PREFIXES):
+        return None
+    return text
+
+
+def is_typed_turn(record: dict) -> bool:
+    return typed_text(record) is not None
diff --git a/plugins/session-flow/skills/audit-sessions/reference/canaries.json b/plugins/session-flow/skills/audit-sessions/reference/canaries.json
index 1add4f767d..6cbbe665b6 100644
--- a/plugins/session-flow/skills/audit-sessions/reference/canaries.json
+++ b/plugins/session-flow/skills/audit-sessions/reference/canaries.json
@@ -3,7 +3,7 @@
   "validated_against": "2.1.287",
   "verified": "2026-10-02",
   "recheck": "a Claude Code release that changes the transcript format, or a drift run that reports new or vanished keys",
-  "basis": "collect.py census over this machine's 104 sessions (2.1.275-2.1.287): each key below is on every record of its population in 2.1.284, 2.1.285 and 2.1.287, except effort (99.8%) and turn_duration (30-47% of system records); invariant:usage_split counted 16 groups, all in subagents, none in 2.1.287",
+  "basis": "collect.py census over this machine's 104 sessions (2.1.275-2.1.287): each key below is on every record of its population in 2.1.284, 2.1.285 and 2.1.287, except effort (99.8%) and turn_duration (30-47% of system records); invariant:usage_split counted 16 groups, all in subagents, none in 2.1.287. key_path:user:entrypoint was added 2026-10-04: it decides which sessions sweep sets aside as automated, and its feeds are the four metrics whose pooled and interactive-only medians the 2026-10-03 audit compared (764 sessions), all of which moved; no other metric was compared, so widen the feeds if a later comparison shows one moving; it was on every user record of headless 2.1.288 sessions, not yet checked over this machine's corpus",
   "canaries": [
     {"key": "key_path:assistant:message.id", "record_type": "assistant", "feeds": ["tokens.main", "tokens.sub"]},
     {"key": "usage_key:input_tokens", "record_type": "assistant", "feeds": ["tokens.main", "tokens.sub"]},
@@ -17,6 +17,7 @@
     {"key": "system_subtype:turn_duration", "record_type": "system", "feeds": ["time.active_s"]},
     {"key": "key_path:system:durationMs", "record_type": "system", "population": "system_subtype:turn_duration", "feeds": ["time.active_s"]},
     {"key": "key_path:system:hookInfos[].durationMs", "record_type": "system", "population": "system_subtype:stop_hook_summary", "feeds": ["hooks.stop_ms"]},
-    {"key": "key_path:user:message.content", "record_type": "user", "feeds": ["human.turns", "tools.interrupts", "context.clear_commands"]}
+    {"key": "key_path:user:message.content", "record_type": "user", "feeds": ["human.turns", "tools.interrupts", "context.clear_commands"]},
+    {"key": "key_path:user:entrypoint", "record_type": "user", "feeds": ["tokens.main", "tokens.sub", "time.active_s", "human.turns"]}
   ]
 }
diff --git a/plugins/session-flow/skills/audit-sessions/reference/store-layout.md b/plugins/session-flow/skills/audit-sessions/reference/store-layout.md
index 560c9dbae9..0f719c4c4e 100644
--- a/plugins/session-flow/skills/audit-sessions/reference/store-layout.md
+++ b/plugins/session-flow/skills/audit-sessions/reference/store-layout.md
@@ -43,8 +43,24 @@ D/audit-sessions/
   `audit_sessions_retention_days` days ago (0 keeps every record), and never ingests one already
   outside that window.
 
+- **How a session was launched.** `entrypoints` lists every `entrypoint` value the main transcript
+  carries, or `["unknown"]` when it carries none; a record written before the field existed reads
+  as unknown too. Collect skips a transcript Claude Code set aside as `<session>.orphaned-*.jsonl`
+  and counts it in `skipped_orphaned`, so it never becomes a second session. A record an earlier
+  collector stored for such a transcript is deleted on the next collect, whether or not the
+  transcript still exists and whatever `--session` selects, and counted in `purged_orphaned`.
+
 ## Reports
 
+- **Interactive sessions only.** Sweep sets aside a session whose entrypoints are all headless or
+  Agent SDK ones (the set is `AUTOMATED_ENTRYPOINTS` in `scripts/census.py`) and reports how many;
+  a session with no known entrypoint stays in and is counted as unclassified. The drift check still
+  reads every record. A headless run of real work is set aside too.
+  - **Pointer**: the transcript key is undocumented; its values match the `app.entrypoint`
+    attribute in <https://code.claude.com/docs/en/monitoring-usage#standard-attributes>.
+  - **As of**: 2026-10-04
+  - **Recheck trigger**: that row adds an SDK entrypoint, or a drift run reports
+    `key_path:user:entrypoint` as a lost canary.
 - `sweep.py --write-report` writes one JSON and one markdown report per run, named by a UTC stamp
   with microseconds, and appends one line to that scope's `history.jsonl`: the stamp, scope, session
   count and each metric's value. The newest 20 report pairs per scope are kept; `history.jsonl` is
diff --git a/plugins/session-flow/skills/audit-sessions/reference/sweep-rules.json b/plugins/session-flow/skills/audit-sessions/reference/sweep-rules.json
index 7ed29d9868..b127eff3ab 100644
--- a/plugins/session-flow/skills/audit-sessions/reference/sweep-rules.json
+++ b/plugins/session-flow/skills/audit-sessions/reference/sweep-rules.json
@@ -78,11 +78,11 @@
       "metric": "tools.interrupts",
       "lens": "interruption",
       "unit": "count",
-      "threshold": 0,
+      "threshold": 1,
       "route": "retro-codify",
       "suggested_skill": "/session-flow:retro",
       "confidence": "measured",
-      "basis": "main-session interrupts per session; spike median 0, p90 0, max 2, 3 of 90 sessions nonzero"
+      "basis": "main-session interrupts per session; spike median 0, p90 0, max 2, 3 of 90 sessions nonzero. Raised from 0 to 1 by operator decision (#6076): the first full store (2026-10-03, 764 sessions) had p90 1, and 57 of the 81 sessions threshold 0 flagged had exactly one interrupt; threshold 1 flags 24"
     },
     {
       "metric": "tools.denials",
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/census.py b/plugins/session-flow/skills/audit-sessions/scripts/census.py
index 5c1a44585c..849affc537 100644
--- a/plugins/session-flow/skills/audit-sessions/scripts/census.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/census.py
@@ -17,7 +17,8 @@
 - `unknown-record-type`: a record type the collector's reader does not know, with its count.
 
 Only versions holding `min_count` records take part, and the `unknown` version bucket (records
-without a `version`) never does. The drift defaults and `write_atomic` live here too, so both
+without a `version`) never does. The drift defaults, `session_class` (a record's stored
+`entrypoints` read as automated, interactive or unknown) and `write_atomic` live here too, so both
 scripts share one store layer without `sweep.py` importing `collect.py` and its reader.
 Stdlib only; Python 3.10+.
 """
@@ -48,6 +49,12 @@
     }
 )
 UNKNOWN_VERSION = "unknown"
+UNKNOWN_ENTRYPOINT = "unknown"
+# Transcript `entrypoint` values of headless (`claude -p`) and Agent SDK runs. The transcript key is
+# undocumented; its values match the `app.entrypoint` telemetry attribute. Pointer:
+# https://code.claude.com/docs/en/monitoring-usage#standard-attributes (as of 2026-10-04; recheck
+# when that row adds an SDK entrypoint or a drift run reports the entrypoint canary lost).
+AUTOMATED_ENTRYPOINTS = frozenset({"sdk-cli", "sdk-ts", "sdk-py"})
 DEFAULT_MIN_COUNT = 20
 DEFAULT_VERSIONS = 3
 # Sessions the newest version needs before a missing canary counts as lost.
@@ -105,6 +112,18 @@ def load_records(data_dir: Path) -> tuple[list[dict], int]:
     return records, skipped
 
 
+def session_class(record: dict) -> str:
+    """`automated` when every stored entrypoint is an SDK one, `unknown` when none is known, else `interactive`.
+
+    A session resumed interactively after a headless start carries both and counts as interactive.
+    """
+    values = record.get("entrypoints")
+    known = [v for v in values if isinstance(v, str) and v != UNKNOWN_ENTRYPOINT] if isinstance(values, list) else []
+    if not known:
+        return "unknown"
+    return "automated" if all(v in AUTOMATED_ENTRYPOINTS for v in known) else "interactive"
+
+
 def load_canaries(path: Path) -> dict:
     canaries = json.loads(Path(path).read_text(encoding="utf-8"))
     if not isinstance(canaries, dict) or canaries.get("schema") != CANARY_SCHEMA:
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/collect.py b/plugins/session-flow/skills/audit-sessions/scripts/collect.py
index fbcf0ea7c9..f3ba3fa5b9 100755
--- a/plugins/session-flow/skills/audit-sessions/scripts/collect.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/collect.py
@@ -8,7 +8,9 @@
     collect.py drift --data-dir D [--min-count N] [--versions N] [--session-floor N] [--canaries FILE]
 
 Writes one `session-record/v1` file per main session (the main transcript plus its subagents)
-under `D/audit-sessions/store/v1/`, the machine-wide store `sweep.py` reads. A session whose
+under `D/audit-sessions/store/v1/`, the machine-wide store `sweep.py` reads; a transcript Claude
+Code set aside (`<session>.orphaned-*.jsonl`) is skipped and counted, not ingested, and a record an
+earlier collector stored for one is deleted. A session whose
 fingerprint matches its stored record is skipped, unless that record was written by another
 collector version, under other excerpt limits, or with redaction failing closed where it now
 works or the reverse. With a retention window, records of sessions
@@ -100,6 +102,10 @@
 TEXT_CAP_FLOOR = 4096
 TOO_LONG = "<too-long>"
 SUPPRESSED = "<suppressed>"
+# `<session>.orphaned-<timestamp>-<suffix>.jsonl` is an earlier transcript Claude Code set aside, not
+# a second session. Pointer: https://code.claude.com/docs/en/claude-directory#cleaned-up-automatically
+# (as of 2026-10-04; recheck when that table renames the set-aside transcript).
+ORPHANED = ".orphaned-"
 
 
 def emit(status: str, summary: str, data: dict, code: int, schema: str = SCHEMA) -> int:
@@ -263,6 +269,7 @@ def __init__(self, excerpt_words: int) -> None:
         self.timestamps: list[float] = []
         self.human_timestamps: list[float] = []
         self.versions: set[str] = set()
+        self.entrypoints: set[str] = set()
         self.branches: set[str] = set()
         self.prs: set[tuple[str, int]] = set()
         self.permission_modes: Counter = Counter()
@@ -299,6 +306,8 @@ def add(self, record: dict, side: str) -> None:
             for field, values in (("version", self.versions), ("gitBranch", self.branches)):
                 if isinstance(record.get(field), str) and record[field]:
                     values.add(record[field])
+            if isinstance(record.get("entrypoint"), str) and record["entrypoint"]:
+                self.entrypoints.add(_label(record["entrypoint"]))
         handler = getattr(self, "_" + kind.replace("-", "_"), None)
         if handler is not None:
             handler(record, side, ts)
@@ -379,8 +388,8 @@ def _user(self, record: dict, side: str, ts: float | None) -> None:
             self.slash[command.group(1)] += 1
             if command.group(1) == "clear" and self.human_turns == 0 and not self.messages["main"]:
                 self.started_with_clear = True
-        elif transcript_reader.is_typed_turn(record):
-            self._typed_turn(record, text, ts)
+        elif (typed := transcript_reader.typed_text(record)) is not None:
+            self._typed_turn(record, typed, ts)
 
     def _typed_turn(self, record: dict, text: str, ts: float | None) -> None:
         index = self.human_turns
@@ -641,6 +650,7 @@ def build_record(
         "repo_identity": repo_identity,
         "worktree": worktree,
         "cc_versions": sorted(scan.versions, key=census.version_key),
+        "entrypoints": sorted(scan.entrypoints) or [census.UNKNOWN_ENTRYPOINT],
         "time": scan.time_block(),
         "models": {side: dict(counts) for side, counts in scan.models.items()},
         "effort": {side: dict(counts) for side, counts in scan.effort.items()},
@@ -773,10 +783,22 @@ def cmd_collect(args: argparse.Namespace) -> int:
     index = load_store(store)
     # A record stored under other settings is re-ingested, so a lowered excerpt limit reaches old records.
     policy = (version, {"chars": args.excerpt_chars, "words": args.excerpt_words}, redactor.fail_closed)
-    scanned = ingested = skipped = expired = too_long = 0
+    scanned = ingested = skipped = expired = orphaned = purged = too_long = 0
     failed: list[dict] = []
     unknown_types: Counter = Counter()
+    # Earlier collectors stored a set-aside transcript as a session, and its source may since be gone.
+    for path in [p for p in index if ORPHANED in p.stem]:
+        try:
+            path.unlink(missing_ok=True)
+        except OSError as exc:
+            failed.append({"session_id": path.stem, "reason": str(exc)})
+            continue
+        del index[path]
+        purged += 1
     for main in sorted(root.glob("*/*.jsonl")):
+        if ORPHANED in main.stem:
+            orphaned += 1
+            continue
         if wanted and main.stem not in wanted:
             continue
         try:
@@ -831,6 +853,8 @@ def cmd_collect(args: argparse.Namespace) -> int:
         "ingested": ingested,
         "skipped_unchanged": skipped,
         "skipped_expired": expired,
+        "skipped_orphaned": orphaned,
+        "purged_orphaned": purged,
         "failed": failed,
         "pruned": pruned,
         "store_records": sum(1 for _ in store.glob("p-*/*.json")),
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/sweep.py b/plugins/session-flow/skills/audit-sessions/scripts/sweep.py
index 2a14827412..143e2bd6ef 100755
--- a/plugins/session-flow/skills/audit-sessions/scripts/sweep.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/sweep.py
@@ -9,7 +9,10 @@
 
 Reads only the `session-record/v1` files `collect.py` wrote; it never opens a transcript. Each
 rule in `reference/sweep-rules.json` names one metric: per-session metrics report their median and
-raise one finding listing every session above the threshold; a null threshold reports only. A
+raise one finding listing every session above the threshold; a null threshold reports only.
+Metrics and findings cover interactive sessions: a session whose stored entrypoints are all
+headless or Agent SDK ones (`census.session_class`) is set aside and counted, one with no known
+entrypoint is kept and counted, and the drift check reads every record. A
 metric fed by a lost drift canary (census.py) is withheld. Prints one JSON envelope (or markdown)
 on stdout; exit 0 pass, 1 warning (a metric degraded), 2 error. Stdlib only; Python 3.10+.
 """
@@ -23,6 +26,7 @@
 import re
 import statistics
 import sys
+from collections import Counter
 from datetime import date, datetime, timezone
 from pathlib import Path
 
@@ -228,7 +232,14 @@ def render_md(data: dict) -> str:
     window = data["window"]
     span = f"{window['since'] or 'start'} to {window['until'] or 'now'}"
     versions = ", ".join(window["cc_versions"]) or "none"
-    lines = [f"# Session audit ({data['scope']})", "", f"Sessions: {window['sessions']} ({span}); Claude Code {versions}", ""]
+    sessions = f"Sessions: {window['sessions']} ({span}); Claude Code {versions}"
+    if window.get("automated_set_aside"):
+        n = window["automated_set_aside"]
+        sessions += f"; {n} automated session{'s' * (n != 1)} set aside"
+    if window.get("unclassified"):
+        n = window["unclassified"]
+        sessions += f"; {n} session{'s' * (n != 1)} without an entrypoint kept"
+    lines = [f"# Session audit ({data['scope']})", "", sessions, ""]
     lines += ["## Metrics", "", "| Metric | Value (per-session median) | Unit | n |", "|---|---|---|---|"]
     for metric, row in data["metrics"].items():
         value = "unavailable (canary lost)" if row.get("degraded") else _fmt(row["value"])
@@ -339,11 +350,20 @@ def main(argv: list[str] | None = None) -> int:
     else:
         scope, key = "machine", "machine"
     records = [r for r in records if in_window(r, args.since, args.until)]
-    window = {"since": args.since and args.since.isoformat(), "until": args.until and args.until.isoformat()}
+    classes = Counter(census.session_class(r) for r in records)
+    records = [r for r in records if census.session_class(r) != "automated"]
+    window = {
+        "since": args.since and args.since.isoformat(),
+        "until": args.until and args.until.isoformat(),
+        "automated_set_aside": classes["automated"],
+        "unclassified": classes["unknown"],
+    }
     fed = {metric for canary in canaries for metric in canary["feeds"]}
     catalog = {name.lstrip("/") for name in args.catalog} if args.catalog else None
     data = build_data(records, rules, drift, fed, scope, window, catalog)
     summary = f"{len(records)} sessions, {len(data['findings'])} findings"
+    if classes["automated"]:
+        summary += f"; {classes['automated']} automated set aside"
     status, code = "pass", 0
     if drift["degraded_metrics"]:
         status, code = "warning", 1
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_collect.py b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_collect.py
index 1b8bee28a9..3d094785d3 100644
--- a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_collect.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_collect.py
@@ -603,6 +603,89 @@ def test_titles_and_paths_outside_cwd_are_redacted(data_dir, multi):
     assert rec["edits"]["by_relpath"] == {"~/notes.md": 1}
 
 
+def write_records(project: Path, sid: str, lines: list[dict]) -> Path:
+    project.mkdir(parents=True, exist_ok=True)
+    path = project / f"{sid}.jsonl"
+    path.write_text("".join(json.dumps(r) + "\n" for r in lines), encoding="utf-8", newline="\n")
+    return path
+
+
+def turn(n: int, text: str, **fields) -> dict:
+    return {"type": "user", "uuid": f"u{n}", "timestamp": f"2026-09-25T10:{n:02d}:30Z",
+            "message": {"role": "user", "content": text}, **fields}
+
+
+def reply(n: int, **fields) -> dict:
+    return {"type": "assistant", "uuid": f"a{n}", "timestamp": f"2026-09-25T10:{n:02d}:00Z",
+            "message": {"id": f"m{n}", "model": "claude-opus-5-5", "content": [{"type": "text", "text": "ok"}]}, **fields}
+
+
+def test_records_carry_every_entrypoint_or_an_explicit_unknown(data_dir, tmp_path):
+    project = tmp_path / "projects" / "proj-e"
+    write_records(project, "e-cli", [turn(0, "hi", promptSource="typed", entrypoint="cli"), reply(1, entrypoint="cli")])
+    write_records(project, "e-sdk", [turn(0, "probe", promptSource="sdk", entrypoint="sdk-cli")])
+    write_records(project, "e-mixed", [turn(0, "go", entrypoint="sdk-cli"), reply(1, entrypoint="claude-desktop")])
+    write_records(project, "e-none", [turn(0, "hi", promptSource="typed")])
+    write_records(project, "e-odd", [turn(0, "hi", entrypoint="free text, not a name")])
+    assert collect(data_dir, tmp_path / "projects").returncode == 0
+    stored = {sid: rec["entrypoints"] for sid, rec in records(data_dir).items()}
+    assert stored == {
+        "e-cli": ["cli"],
+        "e-sdk": ["sdk-cli"],
+        "e-mixed": ["claude-desktop", "sdk-cli"],
+        "e-none": ["unknown"],
+        "e-odd": ["<other>"],
+    }
+
+
+def test_desktop_prompt_after_a_reminder_is_a_typed_turn_without_it(data_dir, tmp_path):
+    reminder = "<system-reminder>\nToday's date is 2026-10-04.\n</system-reminder>\n"
+    desktop = {"origin": {"kind": "human"}, "promptSource": "sdk", "entrypoint": "claude-desktop"}
+    write_records(
+        tmp_path / "projects" / "proj-d",
+        "d-1",
+        [turn(0, "start the refactor", **desktop), reply(1), turn(1, reminder + "retry, keep using sonnet", **desktop)],
+    )
+    assert collect(data_dir, tmp_path / "projects").returncode == 0
+    human = records(data_dir)["d-1"]["human"]
+    assert human["turns"] == 2
+    assert [(t["words"], t["excerpt"]) for t in human["flagged"]] == [(4, "retry, keep using sonnet")]
+
+
+def test_orphaned_transcript_is_not_a_second_session(data_dir, tmp_path):
+    project = tmp_path / "projects" / "proj-o"
+    write_records(project, "gold-0009", [turn(0, "current", promptSource="typed")])
+    write_records(project, "gold-0009.orphaned-1759500000000-ab12", [turn(0, "set aside", promptSource="typed")])
+    result = collect(data_dir, tmp_path / "projects")
+    assert result.returncode == 0, result.stdout + result.stderr
+    data = envelope(result)["data"]
+    assert (data["scanned"], data["ingested"], data["skipped_orphaned"], data["store_records"]) == (1, 1, 1, 1)
+    assert set(records(data_dir)) == {"gold-0009"}
+
+
+@pytest.mark.parametrize("source_kept", [True, False])
+def test_orphaned_record_an_earlier_collector_stored_is_purged(data_dir, tmp_path, source_kept):
+    project = tmp_path / "projects" / "proj-o"
+    write_records(project, "gold-0010", [turn(0, "current", promptSource="typed")])
+    assert collect(data_dir, tmp_path / "projects").returncode == 0
+    normal = next((data_dir / "audit-sessions" / "store" / "v1" / "sessions").glob("p-*/gold-0010.json"))
+    before = normal.read_bytes()
+    # The record an earlier collector wrote for the set-aside transcript, beside the real one.
+    orphan_id = "gold-0010.orphaned-1759500000000-cd34"
+    stale = json.loads(before)
+    stale["session_id"] = orphan_id
+    (normal.parent / f"{orphan_id}.json").write_text(json.dumps(stale), encoding="utf-8")
+    if source_kept:
+        write_records(project, orphan_id, [turn(0, "set aside", promptSource="typed")])
+    result = collect(data_dir, tmp_path / "projects")
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert set(records(data_dir)) == {"gold-0010"}
+    assert normal.read_bytes() == before
+    data = envelope(result)["data"]
+    assert (data["purged_orphaned"], data["skipped_orphaned"], data["store_records"]) == (1, int(source_kept), 1)
+    assert envelope(collect(data_dir, tmp_path / "projects"))["data"]["purged_orphaned"] == 0
+
+
 # resolve_bash mirrors hooks/exec-bash.mjs; the filesystem is injected, as the JS tests do.
 WIN_ENV = {"ProgramFiles": "C:\\Program Files", "PATH": "C:\\Windows\\System32;\"D:\\tools\";C:\\Program Files\\WindowsApps"}
 
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_drift.py b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_drift.py
index 054a7d2779..da7280e175 100644
--- a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_drift.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_drift.py
@@ -341,3 +341,10 @@ def test_bundled_canaries_are_well_formed():
         assert canary["key"].split(":", 1)[0] in census.SECTIONS
         assert canary["feeds"]
         assert canary.get("expect", "present") in ("present", "absent")
+
+
+def test_bundled_canaries_watch_the_entrypoint_that_sets_the_sweep_population():
+    bundled = census.load_canaries(census.BUNDLED_CANARIES)["canaries"]
+    (canary,) = [c for c in bundled if c["key"] == "key_path:user:entrypoint"]
+    assert canary["record_type"] == "user"
+    assert {"tokens.main", "tokens.sub", "time.active_s", "human.turns"} <= set(canary["feeds"])
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_sweep.py b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_sweep.py
index 03cccaa68f..e6244d0cd2 100644
--- a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_sweep.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_sweep.py
@@ -43,6 +43,7 @@ def record(session_id: str, **values) -> dict:
         "subagents": {"count": v["subagents.count"]},
         "stop_hooks": {"ms_p90": v["hooks.stop_ms"]},
         "census": v.get("census", {}),
+        **({"entrypoints": v["entrypoints"]} if "entrypoints" in v else {}),
     }
 
 
@@ -113,8 +114,8 @@ def test_finding_names_outlier_sessions_worst_first(data_dir):
     write_store(
         data_dir,
         record("calm", **{"human.turns": 14}),
-        record("busy", **{"human.turns": 15, "tools.interrupts": 1}),
-        record("worst", **{"human.turns": 40, "tools.interrupts": 2}),
+        record("busy", **{"human.turns": 15, "tools.interrupts": 2}),
+        record("worst", **{"human.turns": 40, "tools.interrupts": 3}),
     )
     env = envelope(sweep(data_dir))
     turns = findings_by_metric(env)["human.turns"]
@@ -127,6 +128,57 @@ def test_finding_names_outlier_sessions_worst_first(data_dir):
     assert findings_by_metric(env)["tools.interrupts"]["evidence"] == [{"session_id": "worst"}, {"session_id": "busy"}]
 
 
+def test_one_interrupt_is_not_a_finding_and_two_are(data_dir):
+    write_store(data_dir, record("one", **{"tools.interrupts": 1}))
+    assert "tools.interrupts" not in findings_by_metric(envelope(sweep(data_dir)))
+    write_store(data_dir, record("two", **{"tools.interrupts": 2}))
+    finding = findings_by_metric(envelope(sweep(data_dir)))["tools.interrupts"]
+    assert finding["evidence"] == [{"session_id": "two"}] and finding["threshold"] == 1
+
+
+def test_automated_sessions_are_set_aside_from_medians_and_findings(data_dir):
+    write_store(
+        data_dir,
+        record("cli", entrypoints=["cli"], **{"tokens.main": 10, "human.turns": 3}),
+        record("headless", entrypoints=["sdk-cli"], **{"tokens.main": 10**7, "tools.interrupts": 5}),
+    )
+    result = sweep(data_dir)
+    assert result.returncode == 0, result.stdout + result.stderr
+    env = envelope(result)
+    data = env["data"]
+    assert data["metrics"]["tokens.main"] == {"value": 10, "unit": "tokens", "n": 1}
+    assert data["metrics"]["human.turns"]["value"] == 3
+    assert data["findings"] == []
+    assert (data["window"]["sessions"], data["window"]["automated_set_aside"]) == (1, 1)
+    assert "1 automated set aside" in env["summary"]
+    md = sweep(data_dir, "--format", "md").stdout
+    assert "Sessions: 1 (start to now)" in md and "1 automated session set aside" in md
+
+
+def test_session_class_reads_every_entrypoint_and_keeps_unknowns(data_dir):
+    write_store(
+        data_dir,
+        record("no-field", **{"tokens.main": 1}),
+        record("unknown", entrypoints=["unknown"], **{"tokens.main": 2}),
+        record("resumed-in-desktop", entrypoints=["claude-desktop", "sdk-cli"], **{"tokens.main": 3}),
+        record("agent-sdk", entrypoints=["sdk-py", "sdk-ts"], **{"tokens.main": 10**7}),
+    )
+    data = envelope(sweep(data_dir))["data"]
+    assert data["metrics"]["tokens.main"]["n"] == 3
+    window = data["window"]
+    assert (window["sessions"], window["automated_set_aside"], window["unclassified"]) == (3, 1, 2)
+    md = sweep(data_dir, "--format", "md").stdout
+    assert "2 sessions without an entrypoint kept" in md
+
+
+def test_drift_still_reads_automated_sessions(data_dir):
+    headless = record("headless", entrypoints=["sdk-cli"])
+    headless["unknown"] = {"record_types": {"relocated": 1}}
+    write_store(data_dir, record("cli", entrypoints=["cli"]), headless)
+    data = envelope(sweep(data_dir))["data"]
+    assert data["drift"]["counts"]["unknown-record-type"] == 1
+
+
 def test_impact_sums_excess_over_threshold_in_its_unit(data_dir):
     over = RULE["tokens.main"]["threshold"]
     write_store(data_dir, record("a", **{"tokens.main": over + 100}), record("b", **{"tokens.main": over + 50}))
@@ -155,7 +207,7 @@ def test_metrics_without_a_canary_are_listed_unchecked(data_dir):
 
 
 def test_suggested_skill_absent_from_catalog_renders_not_installed(data_dir):
-    write_store(data_dir, record("s1", **{"tools.interrupts": 1, "tokens.sub": 10**6}))
+    write_store(data_dir, record("s1", **{"tools.interrupts": 2, "tokens.sub": 10**6}))
     plain = findings_by_metric(envelope(sweep(data_dir)))
     assert plain["tokens.sub"]["suggested_skill"] == "/harness-ops:observability"
     env = envelope(sweep(data_dir, "--catalog", "session-flow:retro"))
@@ -169,7 +221,7 @@ def test_suggested_skill_absent_from_catalog_renders_not_installed(data_dir):
 
 
 def test_catalog_entries_match_with_or_without_leading_slash(data_dir):
-    write_store(data_dir, record("s1", **{"tools.interrupts": 1}))
+    write_store(data_dir, record("s1", **{"tools.interrupts": 2}))
     found = findings_by_metric(envelope(sweep(data_dir, "--catalog", "/session-flow:retro")))
     assert found["tools.interrupts"]["suggested_skill"] == "/session-flow:retro"
 
@@ -242,7 +294,7 @@ def test_md_renders_metrics_and_findings_from_the_same_data(data_dir):
 
 
 def test_md_names_each_findings_sessions_worst_first(data_dir):
-    write_store(data_dir, record("s-low", **{"tools.interrupts": 1}), record("s-high", **{"tools.interrupts": 3}), record("s-none"))
+    write_store(data_dir, record("s-low", **{"tools.interrupts": 2}), record("s-high", **{"tools.interrupts": 3}), record("s-none"))
     findings = sweep(data_dir, "--format", "md").stdout.split("## Findings", 1)[1].split("## Drift", 1)[0]
     line = next(row for row in findings.splitlines() if row.startswith("- `tools.interrupts`"))
     assert line.index("s-high") < line.index("s-low") and "s-none" not in line
@@ -273,8 +325,8 @@ def test_write_report_writes_a_pair_appends_history_and_keeps_20(data_dir):
 def test_scope_project_filters_on_identity_and_keys_reports(data_dir):
     write_store(
         data_dir,
-        record("mine", **{"identity": "github.com/o/r", "tools.interrupts": 1}),
-        record("theirs", **{"identity": "github.com/o/other", "tools.interrupts": 1}),
+        record("mine", **{"identity": "github.com/o/r", "tools.interrupts": 2}),
+        record("theirs", **{"identity": "github.com/o/other", "tools.interrupts": 2}),
     )
     result = sweep(data_dir, "--scope", "project", "--state-key", "github.com/o/r/abcd1234", "--write-report")
     assert result.returncode == 0, result.stdout + result.stderr
diff --git a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_transcript_reader.py b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_transcript_reader.py
index 0cc5afb538..5ad67eb727 100644
--- a/plugins/session-flow/skills/audit-sessions/scripts/tests/test_transcript_reader.py
+++ b/plugins/session-flow/skills/audit-sessions/scripts/tests/test_transcript_reader.py
@@ -292,6 +292,56 @@ def test_injected_prefixes_are_not_typed_turns(prefix):
     assert transcript_reader.is_typed_turn(user([{"type": "text", "text": f"  {prefix}"}])) is False
 
 
+REMINDER = "<system-reminder>\nToday's date is 2026-10-04.\n</system-reminder>\n"
+
+
+def desktop(content: object, **fields) -> dict:
+    """A Claude Desktop user record: typed by the person, yet written with promptSource sdk."""
+    return user(content, origin={"kind": "human"}, promptSource="sdk", **fields)
+
+
+@pytest.mark.parametrize(
+    ("record", "text"),
+    [
+        (desktop("retry, keep using sonnet"), "retry, keep using sonnet"),
+        (desktop(REMINDER + "retry, keep using sonnet"), "retry, keep using sonnet"),
+        (desktop([{"type": "text", "text": REMINDER}, {"type": "text", "text": "retry"}]), "retry"),
+    ],
+    ids=["plain", "leading-reminder", "reminder-block"],
+)
+def test_desktop_human_prompts_are_typed_turns(record, text):
+    assert transcript_reader.is_typed_turn(record) is True
+    assert transcript_reader.typed_text(record) == text
+
+
+@pytest.mark.parametrize(
+    "record",
+    [
+        desktop(REMINDER),
+        desktop(REMINDER + REMINDER + "retry"),
+        desktop(REMINDER + "<command-name>/clear</command-name>"),
+        desktop(REMINDER + "[Request interrupted by user]"),
+        desktop("<system-reminder>never closed retry"),
+        user("retry", promptSource="sdk"),
+        user(REMINDER + "retry"),
+        user(REMINDER + "retry", promptSource="typed"),
+    ],
+    ids=[
+        "reminder-only",
+        "second-reminder",
+        "reminder-then-command",
+        "reminder-then-interrupt",
+        "unclosed-reminder",
+        "sdk-without-origin",
+        "cli-reminder-led",
+        "cli-typed-reminder-led",
+    ],
+)
+def test_reminder_strip_and_sdk_source_need_a_human_origin(record):
+    assert transcript_reader.is_typed_turn(record) is False
+    assert transcript_reader.typed_text(record) is None
+
+
 @pytest.mark.parametrize(
     ("record", "typed"),
     [