From d9b13c84a5c9efb2fec4bf69d94e32ddbb9fd65a Mon Sep 17 00:00:00 2001
From: Makisuo <minetoaster12@gmail.com>
Date: Fri, 2 Oct 2026 00:50:01 +0200
Subject: [PATCH 1/4] feat(cli): scoped `maple delete` for the local store

Deleting from traces left about 20 derived tables holding the rows, so
`maple reset` was the only clean option. `maple delete --service <name>
[--env <env>] [--before <time>]` removes one service's telemetry from the
raw tables and every derived table and rollup.

Every schema table is classified in TABLE_DELETE_PLAN (filter, rebuild
from surviving source rows, or excluded) and validated against the
schema manifest; a test fails when a table is added unclassified. The
delete runs server-side behind the maintenance token inside the
admission gate, is journaled so a partial run resumes, and marks the
store changed so the next checkpoint refresh runs. Local-only.
---
 apps/cli/src/cli.ts                  |   2 +
 apps/cli/src/commands/delete.ts      | 155 +++++++
 apps/cli/src/commands/server-args.ts |   8 +
 apps/cli/src/commands/server.ts      |  11 +-
 apps/cli/src/server/scoped-delete.ts | 626 +++++++++++++++++++++++++++
 apps/cli/src/server/serve.ts         |  60 ++-
 apps/cli/test/delete-command.test.ts |  53 +++
 apps/cli/test/scoped-delete.test.ts  | 208 +++++++++
 apps/cli/test/server-args.test.ts    |   9 +
 docs/local-mode.md                   |  85 +++-
 docs/local-telemetry-archives.md     |   3 +
 11 files changed, 1209 insertions(+), 11 deletions(-)
 create mode 100644 apps/cli/src/commands/delete.ts
 create mode 100644 apps/cli/src/server/scoped-delete.ts
 create mode 100644 apps/cli/test/delete-command.test.ts
 create mode 100644 apps/cli/test/scoped-delete.test.ts

diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts
index 8039780e29..91a44baa6d 100644
--- a/apps/cli/src/cli.ts
+++ b/apps/cli/src/cli.ts
@@ -12,6 +12,7 @@ import { use } from "./commands/config"
 import { start, stop, reset, checkpoint, restore } from "./commands/server"
 import { schema } from "./commands/schema"
 import { archive } from "./commands/archive"
+import { deleteCommand } from "./commands/delete"
 import { update } from "./commands/update"
 import { OutputFormatSetting } from "./lib/output"
 
@@ -48,6 +49,7 @@ export const cli = Command.make("maple").pipe(
 		reset,
 		checkpoint,
 		restore,
+		deleteCommand,
 		schema,
 		// Parquet archives (local mode)
 		archive,
diff --git a/apps/cli/src/commands/delete.ts b/apps/cli/src/commands/delete.ts
new file mode 100644
index 0000000000..0c92999906
--- /dev/null
+++ b/apps/cli/src/commands/delete.ts
@@ -0,0 +1,155 @@
+import { Effect, Option, Schema } from "effect"
+import * as Command from "effect/unstable/cli/Command"
+import * as Flag from "effect/unstable/cli/Flag"
+import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
+import { readFileSync } from "node:fs"
+import { Mode } from "../core/mode"
+import { parseTimestampMs, sinceToMs } from "../core/time"
+import { CliUsageError } from "../lib/errors"
+import { bold, dim, green } from "../lib/style"
+import { maintenanceTokenPath } from "../server/archives/retention"
+import { ScopedDeleteReport, type ScopedDeleteRequest } from "../server/scoped-delete"
+import { jsonFormatRequested, writeJson } from "./json-output"
+import { LocalStatus, prettyPath } from "./server-args"
+
+const LOCAL_ONLY = "maple delete removes data from a local store only"
+
+/** Resolve the local server to delete from; remote mode is refused, never forwarded. */
+export const resolveDeleteBaseUrl = Effect.gen(function* () {
+	if (process.argv.includes("--remote"))
+		return yield* new CliUsageError({
+			message: `${LOCAL_ONLY}; --remote is not supported`,
+			hint: "drop --remote and run it against a `maple start` server",
+		})
+	const resolved = yield* (yield* Mode).resolve
+	if (resolved._tag !== "local")
+		return yield* new CliUsageError({
+			message: `${LOCAL_ONLY}, and the active mode is remote`,
+			hint: "pass --local to target the running `maple start` server",
+		})
+	return resolved.baseUrl.replace(/\/$/, "")
+})
+
+/** `--before` as a relative age (`7d`, `12h`) or an absolute UTC timestamp. */
+export const parseBefore = (raw: string, nowMs: number): number | null => {
+	const age = sinceToMs(raw)
+	return age === null ? parseTimestampMs(raw) : nowMs - age
+}
+
+const failure = (message: string, hint?: string) =>
+	hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint })
+
+const postDelete = (baseUrl: string, body: ScopedDeleteRequest & { readonly dryRun: boolean }) =>
+	Effect.gen(function* () {
+		const client = yield* HttpClient.HttpClient
+		const status = yield* HttpClient.get(`${baseUrl}/local/status`).pipe(
+			Effect.flatMap(HttpClientResponse.filterStatusOk),
+			Effect.flatMap(HttpClientResponse.schemaBodyJson(LocalStatus)),
+			Effect.mapError(() => failure(`no Maple server answered at ${baseUrl}`, "start one with `maple start`")),
+		)
+		// The token sits beside the data dir, so only the machine that owns the store can delete.
+		const tokenPath = maintenanceTokenPath(status.dataDir)
+		const token = yield* Effect.try({
+			try: () => readFileSync(tokenPath, "utf8").trim(),
+			catch: () =>
+				failure(
+					`cannot read the maintenance token at ${prettyPath(tokenPath)}`,
+					"run maple delete as the same user, on the machine that runs `maple start`",
+				),
+		})
+		const response = yield* client
+			.execute(
+				HttpClientRequest.post(`${baseUrl}/local/maintenance/delete`).pipe(
+					HttpClientRequest.setHeader("x-maple-maintenance-token", token),
+					HttpClientRequest.bodyText(JSON.stringify(body), "application/json"),
+				),
+			)
+			.pipe(Effect.mapError((error) => failure(`delete request failed: ${error.message}`)))
+		const text = yield* response.text.pipe(Effect.orElseSucceed(() => ""))
+		if (response.status === 404)
+			return yield* failure("this server predates `maple delete`", "update with `maple update` and restart it")
+		if (response.status < 200 || response.status >= 300)
+			return yield* failure(text || `server answered HTTP ${response.status}`)
+		return yield* Schema.decodeUnknownEffect(Schema.fromJsonString(ScopedDeleteReport))(text).pipe(
+			Effect.mapError(() => failure("the server returned an unreadable delete report")),
+		)
+	})
+
+const describeRequest = (request: ScopedDeleteRequest): string =>
+	[
+		`service ${bold(request.service)}`,
+		...(request.env === undefined ? [] : [`env ${bold(request.env === "" ? '""' : request.env)}`]),
+		...(request.beforeMs === undefined ? [] : [`before ${bold(new Date(request.beforeMs).toISOString())}`]),
+	].join(", ")
+
+const renderReport = (report: ScopedDeleteReport): string => {
+	const lines = report.tables
+		.filter((t) => t.strategy !== "excluded" && (t.rows > 0 || (t.skippedHours ?? 0) > 0))
+		.map((t) =>
+			t.strategy === "filter"
+				? `  ${t.table.padEnd(36)} ${t.rows} row(s)`
+				: `  ${t.table.padEnd(36)} ${t.rebuiltHours} hour(s) recomputed` +
+					((t.skippedHours ?? 0) > 0 ? dim(`, ${t.skippedHours} past source retention kept`) : ""),
+		)
+	return lines.length === 0 ? `  ${dim("nothing matched")}\n` : `${lines.join("\n")}\n`
+}
+
+export const deleteCommand = Command.make("delete", {
+	service: Flag.String("service").pipe(Flag.withDescription("service.name whose telemetry to delete")),
+	env: Flag.optional(
+		Flag.String("env").pipe(
+			Flag.withDescription("Only rows whose deployment.environment(.name) equals this value"),
+		),
+	),
+	before: Flag.optional(
+		Flag.String("before").pipe(
+			Flag.withDescription(
+				"Only rows older than this: an age (7d, 12h) or UTC timestamp; floored to the hour",
+			),
+		),
+	),
+	yes: Flag.Boolean("yes").pipe(
+		Flag.withDescription("Delete; without it the command only previews what would go"),
+		Flag.withDefault(false),
+	),
+}).pipe(
+	Command.withDescription(
+		"Delete one service's telemetry from the local store: raw tables and every derived rollup (local only)",
+	),
+	Command.withExamples([
+		{ command: "maple delete --service checkout-pr-42" },
+		{ command: "maple delete --service api --env preview-17 --yes" },
+		{ command: "maple delete --service api --before 7d --yes" },
+	]),
+	Command.withHandler(
+		Effect.fnUntraced(function* (a) {
+			const baseUrl = yield* resolveDeleteBaseUrl
+			const service = a.service.trim()
+			if (service.length === 0) return yield* failure("--service must not be empty")
+			const rawBefore = Option.getOrUndefined(a.before)
+			const beforeMs = rawBefore === undefined ? undefined : parseBefore(rawBefore, Date.now())
+			if (beforeMs === null)
+				return yield* failure(`invalid --before: ${rawBefore}`, "use an age like 7d or a timestamp like 2026-10-01 12:00")
+			const env = Option.getOrUndefined(a.env)
+			const base = { service }
+			const withEnv = env === undefined ? base : { ...base, env }
+			const request = beforeMs === undefined ? withEnv : { ...withEnv, beforeMs: Math.floor(beforeMs) }
+			const report = yield* postDelete(baseUrl, { ...request, dryRun: !a.yes })
+			if (jsonFormatRequested()) yield* writeJson(report)
+			else
+				yield* Effect.sync(() =>
+					process.stderr.write(
+						`${a.yes ? green("✓ deleted") : "would delete"} ${describeRequest(report.request)}\n` +
+							renderReport(report) +
+							(a.yes
+								? dim(
+										"  Checkpoints taken before now still hold these rows until the next refresh; `maple checkpoint` refreshes now.\n",
+									)
+								: ""),
+					),
+				)
+			if (!a.yes && report.tables.some((t) => t.rows > 0))
+				return yield* failure("nothing was deleted (preview)", "re-run with --yes to delete")
+		}),
+	),
+)
diff --git a/apps/cli/src/commands/server-args.ts b/apps/cli/src/commands/server-args.ts
index 807ffade1c..e5c1f8bc02 100644
--- a/apps/cli/src/commands/server-args.ts
+++ b/apps/cli/src/commands/server-args.ts
@@ -99,6 +99,8 @@ export const LocalStatus = Schema.Struct({
 	url: Schema.String,
 	dataDir: Schema.String,
 	lastIngestAtMs: Schema.NullOr(Schema.Number),
+	/** Absent from binaries that predate `maple delete`. */
+	lastDeleteAtMs: Schema.optionalKey(Schema.NullOr(Schema.Number)),
 })
 export type LocalStatus = typeof LocalStatus.Type
 
@@ -133,6 +135,12 @@ export const startedBeforeWrite = (startedAtMs: number, writtenAtMs: number): bo
 export const ingestedSince = (lastIngestAtMs: number | null | undefined, sinceMs: number): boolean =>
 	lastIngestAtMs === undefined || (lastIngestAtMs !== null && lastIngestAtMs >= sinceMs)
 
+/** Did ingest or a scoped delete change the store since `sinceMs`? Unknown status always did. */
+export const storeChangedSince = (status: LocalStatus | undefined, sinceMs: number): boolean =>
+	status === undefined ||
+	ingestedSince(status.lastIngestAtMs, sinceMs) ||
+	(status.lastDeleteAtMs != null && status.lastDeleteAtMs >= sinceMs)
+
 export interface DetachedChildArgs {
 	readonly entry: string | undefined
 	readonly host: string
diff --git a/apps/cli/src/commands/server.ts b/apps/cli/src/commands/server.ts
index 17edfcf16d..5a09e641c7 100644
--- a/apps/cli/src/commands/server.ts
+++ b/apps/cli/src/commands/server.ts
@@ -76,7 +76,7 @@ import {
 	type DirtyStorePolicy,
 	hostedDashboardUrl,
 	hostedUiOrigin,
-	ingestedSince,
+	storeChangedSince,
 	isProcessAlive,
 	LocalStatus,
 	mapleCommand,
@@ -852,8 +852,8 @@ const takeCheckpointQuietly = (
  *
  * `BACKUP` runs on the server's own JS thread and stalls ingest and queries
  * while it runs, so an idle server skips it: once this process has a
- * checkpoint, a tick with nothing ingested since that checkpoint started
- * (`/local/status` `lastIngestAtMs`) has nothing new to protect. Consecutive
+ * checkpoint, a tick with nothing ingested or deleted since that checkpoint
+ * started (`/local/status` `lastIngestAtMs`/`lastDeleteAtMs`) has nothing new to protect. Consecutive
  * failures back the interval off (`checkpointRefreshBackoff`).
  */
 const checkpointRefreshLoop = (
@@ -877,9 +877,8 @@ const checkpointRefreshLoop = (
 			yield* Effect.sleep(backoff.delay)
 			if (lastTakenAtMs !== undefined) {
 				const status = yield* fetchLocalStatus(statusUrl, "5 seconds")
-				const lastIngestAtMs = Option.isSome(status) ? status.value.lastIngestAtMs : undefined
-				if (!ingestedSince(lastIngestAtMs, lastTakenAtMs)) {
-					debugLog("checkpoint refresh skipped", "nothing ingested since the last checkpoint")
+				if (!storeChangedSince(Option.getOrUndefined(status), lastTakenAtMs)) {
+					debugLog("checkpoint refresh skipped", "nothing ingested or deleted since the last checkpoint")
 					continue
 				}
 			}
diff --git a/apps/cli/src/server/scoped-delete.ts b/apps/cli/src/server/scoped-delete.ts
new file mode 100644
index 0000000000..c213d62291
--- /dev/null
+++ b/apps/cli/src/server/scoped-delete.ts
@@ -0,0 +1,626 @@
+// Scoped delete for the local store: one service's telemetry (optionally one
+// deployment.environment and/or everything before an hour) leaves the raw
+// tables and every table derived from them. See docs/local-mode.md.
+
+import { Effect, Option, Schema } from "effect"
+import { existsSync } from "node:fs"
+import type { Chdb } from "./chdb"
+import { decodeJsonEachRow, decodeRowCounts } from "./chdb-rows"
+import { durableJson, durableRemove } from "./durable-files"
+import { readRealFile } from "./local-token"
+import { type LocalSchemaManifest, ttlDaysFromDefinition, viewBody } from "./schema-manifest"
+import { dataDirSidecarPath } from "./store-version"
+
+const HOUR_MS = 3_600_000
+const DAY_MS = 24 * HOUR_MS
+
+type Db = Pick<Chdb, "query" | "exec">
+
+export class ScopedDeleteError extends Schema.TaggedError<ScopedDeleteError>()(
+	"@maple/cli/ScopedDeleteError",
+	{ message: Schema.String, cause: Schema.optionalKey(Schema.Defect()) },
+) {}
+
+/** Where a table's environment lives: a column, or the OTel resource map. */
+export type EnvSource = { readonly column: string } | { readonly resourceAttributes: string }
+
+/** `filter` deletes by column (`env: null`: no env, so `--env` recomputes it), `rebuild`
+ * recomputes affected hours from surviving source rows, `excluded` never holds local telemetry. */
+export type TableDeletePlan =
+	| {
+			readonly strategy: "filter"
+			/** A row matches when any of these columns equals `--service`. */
+			readonly service: ReadonlyArray<string>
+			readonly env: EnvSource | null
+			readonly time: string
+	  }
+	| { readonly strategy: "rebuild"; readonly time: string; readonly reason: string }
+	| { readonly strategy: "excluded"; readonly reason: string }
+
+const RESOURCE_ENV: EnvSource = { resourceAttributes: "ResourceAttributes" }
+const DEPLOYMENT_ENV: EnvSource = { column: "DeploymentEnv" }
+
+const raw = (time: string): TableDeletePlan => ({
+	strategy: "filter",
+	service: ["ServiceName"],
+	env: RESOURCE_ENV,
+	time,
+})
+const derived = (time: string, env: EnvSource | null = DEPLOYMENT_ENV): TableDeletePlan => ({
+	strategy: "filter",
+	service: ["ServiceName"],
+	env,
+	time,
+})
+const edges = (target: string): TableDeletePlan => ({
+	strategy: "filter",
+	service: ["SourceService", target],
+	env: DEPLOYMENT_ENV,
+	time: "Hour",
+})
+const NOT_TELEMETRY = "control-plane history, not telemetry; local mode never writes it"
+const SESSION_DATA = "browser session data; local OTLP ingest never writes it"
+
+/**
+ * Every table in the local schema. `validateDeletePlan` (and its test) fails when
+ * the schema gains a table this map does not classify.
+ */
+/** Delete strategy per table name; keys are checked against the schema at runtime. */
+export type TableDeletePlans = Readonly<Record<string, TableDeletePlan>>
+
+export const TABLE_DELETE_PLAN = {
+	ai_crawler_requests: derived("Timestamp", null),
+	ai_trace_index: derived("Timestamp"),
+	alert_checks: { strategy: "excluded", reason: NOT_TELEMETRY },
+	attribute_keys_hourly: {
+		strategy: "rebuild",
+		time: "Hour",
+		reason: "org-wide attribute discovery; rows have no service column",
+	},
+	attribute_values_hourly: {
+		strategy: "rebuild",
+		time: "Hour",
+		reason: "org-wide attribute discovery; rows have no service column",
+	},
+	audit_log: { strategy: "excluded", reason: NOT_TELEMETRY },
+	error_events: derived("Timestamp"),
+	error_events_by_time: derived("Timestamp"),
+	error_fingerprints_minutely: {
+		strategy: "rebuild",
+		time: "Minute",
+		reason: "ServiceName is an anyLast aggregate, not a key; one row can mix services",
+	},
+	identity_links: { strategy: "excluded", reason: SESSION_DATA },
+	logs: raw("Timestamp"),
+	logs_aggregates_hourly: derived("Hour"),
+	metric_catalog: derived("Hour", null),
+	metrics_exponential_histogram: raw("TimeUnix"),
+	metrics_gauge: raw("TimeUnix"),
+	metrics_histogram: raw("TimeUnix"),
+	metrics_sum: raw("TimeUnix"),
+	product_events: derived("Timestamp", null),
+	service_address_resolutions_hourly: edges("ResolvedTargetService"),
+	service_external_edges_hourly: derived("Hour"),
+	service_map_children: derived("Timestamp"),
+	service_map_db_edges_hourly: derived("Hour"),
+	service_map_db_query_shapes_hourly: derived("Hour"),
+	service_map_edges_hourly: edges("TargetService"),
+	service_map_edges_hourly_ingest: {
+		strategy: "excluded",
+		reason: "Null engine: stores no rows; its view writes service_map_edges_hourly",
+	},
+	service_map_spans: derived("Timestamp"),
+	service_operations_hourly: derived("Hour"),
+	service_operations_minutely: derived("Minute"),
+	service_overview_hourly: derived("Hour"),
+	service_overview_minutely: derived("Minute"),
+	service_overview_spans: derived("Timestamp"),
+	service_platforms_hourly: derived("Hour"),
+	service_usage: derived("Hour", null),
+	session_events: { strategy: "excluded", reason: SESSION_DATA },
+	session_replay_events: { strategy: "excluded", reason: SESSION_DATA },
+	session_replays: { strategy: "excluded", reason: SESSION_DATA },
+	span_metrics_calls_hourly: derived("Hour", null),
+	trace_detail_spans: derived("Timestamp", RESOURCE_ENV),
+	trace_facets_hourly: derived("Hour"),
+	trace_list_mv: derived("Timestamp"),
+	traces: raw("Timestamp"),
+	traces_aggregates_hourly: derived("Hour"),
+} satisfies TableDeletePlans
+
+const lookup =
+	(plans: TableDeletePlans) =>
+	(table: string): TableDeletePlan | undefined =>
+		plans[table]
+/** The plan for any table name, or undefined when the schema has no such table. */
+export const tablePlan = lookup(TABLE_DELETE_PLAN)
+
+// ---------------------------------------------------------------------------
+// Schema graph.
+// ---------------------------------------------------------------------------
+
+interface ViewEdge {
+	readonly name: string
+	readonly target: string
+	readonly sources: ReadonlyArray<string>
+	readonly body: string
+}
+
+const IDENTIFIER = /[A-Za-z_][A-Za-z0-9_]*/g
+
+const materializedViews = (manifest: LocalSchemaManifest): ReadonlyArray<ViewEdge> => {
+	const tables = new Set(manifest.objects.filter((o) => o.kind === "table").map((o) => o.name))
+	return manifest.objects
+		.filter((o) => o.kind === "materialized_view")
+		.map((o) => {
+			const body = viewBody(o.definition) ?? ""
+			const sources = [...body.matchAll(/\bFROM\s+([A-Za-z_][A-Za-z0-9_]*)/gi)]
+				.map((match) => match[1]!)
+				.filter((name) => tables.has(name))
+			return {
+				name: o.name,
+				target: o.definition.match(/\bTO\s+([A-Za-z_][A-Za-z0-9_]*)/i)?.[1] ?? "",
+				sources: [...new Set(sources)],
+				body,
+			}
+		})
+}
+
+const isRecomputable = (plan: TableDeletePlan | undefined): boolean =>
+	plan?.strategy === "rebuild" || (plan?.strategy === "filter" && plan.env === null)
+
+/**
+ * Everything that must hold for the plan to delete exactly the requested rows
+ * from the given schema. Empty means sound; the server refuses otherwise.
+ */
+export const validateDeletePlan = (
+	manifest: LocalSchemaManifest,
+	plan: TableDeletePlans = TABLE_DELETE_PLAN,
+): ReadonlyArray<string> => {
+	const problems: string[] = []
+	const tables = new Map(manifest.objects.filter((o) => o.kind === "table").map((o) => [o.name, o]))
+	const views = materializedViews(manifest)
+	for (const name of tables.keys())
+		if (plan[name] === undefined) problems.push(`${name}: no delete plan for this table`)
+	for (const name of Object.keys(plan))
+		if (!tables.has(name)) problems.push(`${name}: planned but not in the schema`)
+
+	for (const [name, entry] of Object.entries(plan)) {
+		const table = tables.get(name)
+		if (table === undefined || entry.strategy === "excluded") continue
+		const columns = new Set(table.columns.map((c) => c.name))
+		const keyColumns = new Set(table.orderBy?.match(IDENTIFIER) ?? [])
+		// Only a plain MergeTree row is one event; any other engine merges rows by key.
+		const merges = table.engine !== "MergeTree"
+		const needColumn = (column: string, role: string) => {
+			if (!columns.has(column)) problems.push(`${name}: ${role} column ${column} does not exist`)
+			else if (merges && !keyColumns.has(column))
+				problems.push(`${name}: ${role} column ${column} is not in the sorting key of ${table.engine}`)
+		}
+		needColumn(entry.time, "time")
+		if (entry.strategy === "filter") {
+			for (const column of entry.service) needColumn(column, "service")
+			if (entry.env !== null) {
+				if ("column" in entry.env) needColumn(entry.env.column, "env")
+				else if (merges) problems.push(`${name}: a merging engine cannot filter env from a map`)
+				else if (!columns.has(entry.env.resourceAttributes))
+					problems.push(`${name}: env map ${entry.env.resourceAttributes} does not exist`)
+			}
+		}
+		if (!isRecomputable(entry)) continue
+		if (entry.strategy === "filter" && (entry.service.length !== 1 || entry.service[0] !== "ServiceName"))
+			problems.push(`${name}: a recomputed service-scoped table must key on ServiceName alone`)
+		const feeding = views.filter(
+			(view) => view.target === name && view.sources.some((s) => plan[s]?.strategy !== "excluded"),
+		)
+		if (feeding.length === 0) problems.push(`${name}: recomputed but no view writes it from telemetry`)
+		for (const view of feeding) {
+			if (view.sources.length !== 1) {
+				problems.push(`${view.name}: recomputing needs exactly one source table`)
+				continue
+			}
+			const source = view.sources[0]!
+			const sourcePlan = plan[source]
+			if (sourcePlan?.strategy !== "filter" || sourcePlan.env === null)
+				problems.push(`${view.name}: source ${source} must be a filter table with an env dimension`)
+			if (sourceReferences(view.body, source) !== 1)
+				problems.push(`${view.name}: recomputing needs exactly one FROM ${source}`)
+		}
+		if (views.some((view) => view.sources.includes(name)))
+			problems.push(`${name}: recomputed rows would cascade into the views it feeds`)
+	}
+
+	// Rows derived from deletable telemetry must never land in an excluded table.
+	for (const view of views) {
+		if (plan[view.target]?.strategy !== "excluded") continue
+		for (const source of view.sources)
+			if (plan[source] !== undefined && plan[source]!.strategy !== "excluded")
+				problems.push(`${view.name}: writes excluded ${view.target} from ${source}`)
+	}
+	return problems
+}
+
+const fromPattern = (source: string) => new RegExp(`\\bFROM\\s+${source}\\b`, "gi")
+const sourceReferences = (body: string, source: string): number =>
+	[...body.matchAll(fromPattern(source))].length
+
+// ---------------------------------------------------------------------------
+// Request and SQL.
+// ---------------------------------------------------------------------------
+
+export const ScopedDeleteRequest = Schema.Struct({
+	service: Schema.NonEmptyString.check(Schema.isMaxLength(512)),
+	env: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(512))),
+	/** Exclusive cutoff, epoch ms; the server floors it to the UTC hour. */
+	beforeMs: Schema.optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))),
+})
+export type ScopedDeleteRequest = typeof ScopedDeleteRequest.Type
+
+/** Bucketed rollups cannot split an hour, so every table uses the same hour cutoff. */
+export const floorToHour = (ms: number): number => Math.floor(ms / HOUR_MS) * HOUR_MS
+
+const sqlString = (value: string): string => `'${value.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'`
+
+const envExpression = (env: EnvSource): string =>
+	"column" in env
+		? env.column
+		: `coalesce(nullIf(${env.resourceAttributes}['deployment.environment.name'], ''), ${env.resourceAttributes}['deployment.environment'])`
+
+const hourOf = (column: string) => `toUnixTimestamp(toStartOfHour(toDateTime(${column})))`
+
+/** The rows a request names in a filter table, or null when `--env` cannot apply there. */
+const filterPredicate = (
+	entry: Extract<TableDeletePlan, { strategy: "filter" }>,
+	request: ScopedDeleteRequest,
+): string | null => {
+	if (request.env !== undefined && entry.env === null) return null
+	const service = sqlString(request.service)
+	const parts = [`(${entry.service.map((column) => `${column} = ${service}`).join(" OR ")})`]
+	if (request.env !== undefined && entry.env !== null)
+		parts.push(`${envExpression(entry.env)} = ${sqlString(request.env)}`)
+	if (request.beforeMs !== undefined)
+		parts.push(`${entry.time} < toDateTime(${Math.floor(request.beforeMs / 1000)})`)
+	return parts.join(" AND ")
+}
+
+const hourList = (hours: ReadonlyArray<number>) => hours.join(", ")
+
+// ---------------------------------------------------------------------------
+// Execution.
+// ---------------------------------------------------------------------------
+
+export const ScopedDeleteTableReport = Schema.Struct({
+	table: Schema.String,
+	strategy: Schema.Literals(["filter", "rebuild", "excluded"]),
+	/** Filter: rows deleted. Rebuild: rows in the recomputed hours before recomputing. */
+	rows: Schema.Number,
+	rebuiltHours: Schema.optionalKey(Schema.Number),
+	/** Hours older than a source's retention: left as-is because they cannot be recomputed. */
+	skippedHours: Schema.optionalKey(Schema.Number),
+	reason: Schema.optionalKey(Schema.String),
+})
+export type ScopedDeleteTableReport = typeof ScopedDeleteTableReport.Type
+
+export const ScopedDeleteReport = Schema.Struct({
+	request: ScopedDeleteRequest,
+	dryRun: Schema.Boolean,
+	deletedRows: Schema.Number,
+	tables: Schema.Array(ScopedDeleteTableReport),
+})
+export type ScopedDeleteReport = typeof ScopedDeleteReport.Type
+
+interface Rebuild {
+	readonly table: string
+	readonly time: string
+	/** Set when only this service's rows are recomputed (a keyed table under `--env`). */
+	readonly serviceColumn: string | null
+	readonly views: ReadonlyArray<ViewEdge & { readonly source: string }>
+}
+
+interface ResolvedPlan {
+	readonly filters: ReadonlyArray<{ readonly table: string; readonly predicate: string }>
+	readonly rebuilds: ReadonlyArray<Rebuild>
+	readonly excluded: ReadonlyArray<{ readonly table: string; readonly reason: string }>
+}
+
+const resolvePlan = (manifest: LocalSchemaManifest, request: ScopedDeleteRequest): ResolvedPlan => {
+	const views = materializedViews(manifest)
+	const filters: Array<ResolvedPlan["filters"][number]> = []
+	const rebuilds: Rebuild[] = []
+	const excluded: Array<ResolvedPlan["excluded"][number]> = []
+	for (const [table, entry] of Object.entries(TABLE_DELETE_PLAN)) {
+		if (entry.strategy === "excluded") {
+			excluded.push({ table, reason: entry.reason })
+			continue
+		}
+		const predicate = entry.strategy === "filter" ? filterPredicate(entry, request) : null
+		if (predicate !== null) {
+			filters.push({ table, predicate })
+			continue
+		}
+		rebuilds.push({
+			table,
+			time: entry.time,
+			serviceColumn: entry.strategy === "filter" ? "ServiceName" : null,
+			views: views
+				.filter((view) => view.target === table)
+				.flatMap((view) => {
+					const source = view.sources[0]
+					return source !== undefined && tablePlan(source)?.strategy === "filter"
+						? [{ ...view, source }]
+						: []
+				}),
+		})
+	}
+	return { filters, rebuilds, excluded }
+}
+
+const run = <A>(label: string, f: () => A) =>
+	Effect.try({
+		try: f,
+		catch: (cause) =>
+			new ScopedDeleteError({
+				message: `${label}: ${cause instanceof Error ? cause.message : String(cause)}`,
+				cause,
+			}),
+	})
+
+const countRows = (db: Db, table: string, predicate: string) =>
+	run(`count ${table}`, () =>
+		Number(
+			decodeRowCounts(
+				db.query(`SELECT toString(count()) AS rowCount FROM ${table} WHERE ${predicate}`),
+			)[0]?.rowCount ?? 0,
+		),
+	)
+
+const decodeHours = decodeJsonEachRow(Schema.Struct({ hour: Schema.String }))
+const decodeNames = decodeJsonEachRow(Schema.Struct({ name: Schema.String }))
+
+const distinctHours = (db: Db, table: string, time: string, predicate: string) =>
+	run(`hours of ${table}`, () =>
+		decodeHours(
+			db.query(`SELECT DISTINCT toString(${hourOf(time)}) AS hour FROM ${table} WHERE ${predicate}`),
+		).map((row) => Number(row.hour)),
+	)
+
+/** Hours (epoch seconds) a rebuild must recompute: the hours its sources lose rows in. */
+const affectedHours = (db: Db, rebuild: Rebuild, request: ScopedDeleteRequest) =>
+	Effect.gen(function* () {
+		const hours = new Set<number>()
+		for (const source of new Set(rebuild.views.map((view) => view.source))) {
+			const entry = tablePlan(source)
+			if (entry?.strategy !== "filter") continue
+			const predicate = filterPredicate(entry, request)
+			if (predicate === null) continue
+			for (const hour of yield* distinctHours(db, source, entry.time, predicate)) hours.add(hour)
+		}
+		return [...hours].sort((a, b) => a - b)
+	})
+
+/**
+ * An hour is recomputable only while every source still holds all of it. TTL
+ * drops whole days at merge time, so keep a day of margin; the bundled TTL is
+ * the shortest a store can have, so this never recomputes from a partial source.
+ */
+const recomputable = (
+	manifest: LocalSchemaManifest,
+	rebuild: Rebuild,
+	hours: ReadonlyArray<number>,
+	nowMs: number,
+) => {
+	const ttls = rebuild.views.map((view) => {
+		const definition = manifest.objects.find((o) => o.name === view.source)?.definition ?? ""
+		return ttlDaysFromDefinition(definition)
+	})
+	const kept: number[] = []
+	const skipped: number[] = []
+	for (const hour of hours) {
+		const covered = ttls.every((days) => days === null || hour * 1000 + days * DAY_MS > nowMs + DAY_MS)
+		;(covered ? kept : skipped).push(hour)
+	}
+	return { kept, skipped }
+}
+
+const rebuildTargetPredicate = (rebuild: Rebuild, request: ScopedDeleteRequest, hours: ReadonlyArray<number>) =>
+	[
+		...(rebuild.serviceColumn === null ? [] : [`${rebuild.serviceColumn} = ${sqlString(request.service)}`]),
+		`${hourOf(rebuild.time)} IN (${hourList(hours)})`,
+	].join(" AND ")
+
+const applyRebuild = (
+	db: Db,
+	manifest: LocalSchemaManifest,
+	rebuild: Rebuild,
+	request: ScopedDeleteRequest,
+	hours: ReadonlyArray<number>,
+) =>
+	Effect.gen(function* () {
+		if (hours.length === 0) return
+		yield* run(`clear ${rebuild.table}`, () =>
+			db.exec(
+				`ALTER TABLE ${rebuild.table} DELETE WHERE ${rebuildTargetPredicate(rebuild, request, hours)} SETTINGS mutations_sync = 2`,
+			),
+		)
+		// One UTC day per INSERT keeps each recompute's GROUP BY bounded.
+		const days = new Map<number, number[]>()
+		for (const hour of hours) {
+			const day = Math.floor(hour / 86_400)
+			days.set(day, [...(days.get(day) ?? []), hour])
+		}
+		for (const view of rebuild.views) {
+			const sourceColumns = manifest.objects
+				.find((o) => o.name === view.source)
+				?.columns.map((c) => c.name)
+				.join(", ")
+			const sourceTime = tablePlan(view.source)
+			if (sourceColumns === undefined || sourceTime?.strategy !== "filter")
+				return yield* new ScopedDeleteError({ message: `${view.name}: unknown source ${view.source}` })
+			// The view's own output columns, so the INSERT matches by name, not position.
+			const columns = yield* run(`columns of ${view.name}`, () =>
+				decodeNames(
+					db.query(
+						`SELECT name FROM system.columns WHERE database = currentDatabase() AND table = ${sqlString(view.name)} ORDER BY position`,
+					),
+				).map((row) => row.name),
+			)
+			if (columns.length === 0)
+				return yield* new ScopedDeleteError({ message: `${view.name}: no columns in system.columns` })
+			for (const dayHours of days.values()) {
+				const scope = [
+					...(rebuild.serviceColumn === null ? [] : [`ServiceName = ${sqlString(request.service)}`]),
+					`${hourOf(sourceTime.time)} IN (${hourList(dayHours)})`,
+				].join(" AND ")
+				const body = view.body.replace(
+					fromPattern(view.source),
+					`FROM (SELECT ${sourceColumns} FROM ${view.source} WHERE ${scope}) AS ${view.source}`,
+				)
+				const list = columns.join(", ")
+				yield* run(`recompute ${rebuild.table} via ${view.name}`, () =>
+					db.exec(`INSERT INTO ${rebuild.table} (${list}) SELECT ${list} FROM (${body})`),
+				)
+			}
+		}
+	})
+
+const applyFilters = (db: Db, filters: ResolvedPlan["filters"]) =>
+	Effect.forEach(
+		filters,
+		({ table, predicate }) =>
+			Effect.gen(function* () {
+				yield* run(`delete from ${table}`, () =>
+					db.exec(`ALTER TABLE ${table} DELETE WHERE ${predicate} SETTINGS mutations_sync = 2`),
+				)
+				const left = yield* countRows(db, table, predicate)
+				if (left !== 0)
+					return yield* new ScopedDeleteError({ message: `${left} matching row(s) remain in ${table}` })
+			}),
+		{ discard: true },
+	)
+
+// ---------------------------------------------------------------------------
+// Journal: written before the first mutation, removed after the last.
+// ---------------------------------------------------------------------------
+
+const PendingDelete = Schema.Struct({
+	formatVersion: Schema.Literal(1),
+	request: ScopedDeleteRequest,
+	rebuildHours: Schema.Record(Schema.String, Schema.Array(Schema.Int)),
+})
+type PendingDelete = typeof PendingDelete.Type
+
+export const pendingDeletePath = (dataDir: string): string =>
+	dataDirSidecarPath(dataDir, "maple-pending-delete.json")
+
+const readPending = (dataDir: string) =>
+	Effect.gen(function* () {
+		const path = pendingDeletePath(dataDir)
+		if (!existsSync(path)) return Option.none<PendingDelete>()
+		const text = yield* run("read pending delete", () => readRealFile(path, "pending delete journal"))
+		return Option.some(
+			yield* Schema.decodeUnknownEffect(Schema.fromJsonString(PendingDelete))(text).pipe(
+				Effect.mapError(
+					(cause) =>
+						new ScopedDeleteError({ message: `malformed pending delete journal ${path}`, cause }),
+				),
+			),
+		)
+	})
+
+const writeFile = (label: string, f: () => Promise<void>) =>
+	Effect.tryPromise({
+		try: f,
+		catch: (cause) => new ScopedDeleteError({ message: `${label}: ${String(cause)}`, cause }),
+	})
+
+const assertPlanSound = (manifest: LocalSchemaManifest) => {
+	const problems = validateDeletePlan(manifest)
+	return problems.length === 0
+		? Effect.void
+		: Effect.fail(
+				new ScopedDeleteError({ message: `scoped delete plan does not match the schema: ${problems.join("; ")}` }),
+			)
+}
+
+/** Apply a journaled request: filters are idempotent, rebuilds take their hours from the journal. */
+const applyPending = (db: Db, manifest: LocalSchemaManifest, dataDir: string, pending: PendingDelete) =>
+	Effect.gen(function* () {
+		const plan = resolvePlan(manifest, pending.request)
+		yield* applyFilters(db, plan.filters)
+		for (const rebuild of plan.rebuilds)
+			yield* applyRebuild(db, manifest, rebuild, pending.request, pending.rebuildHours[rebuild.table] ?? [])
+		yield* writeFile("remove pending delete journal", () => durableRemove(pendingDeletePath(dataDir)))
+	})
+
+/** Finish a delete an earlier failure or crash left journaled. Run before serving. */
+export const resumePendingDelete = (db: Db, manifest: LocalSchemaManifest, dataDir: string) =>
+	Effect.gen(function* () {
+		const pending = yield* readPending(dataDir)
+		if (Option.isNone(pending)) return Option.none<ScopedDeleteRequest>()
+		yield* assertPlanSound(manifest)
+		// A restored checkpoint can bring the rows back; recompute their hours too.
+		const rebuildHours: Record<string, number[]> = {}
+		for (const rebuild of resolvePlan(manifest, pending.value.request).rebuilds) {
+			const current = yield* affectedHours(db, rebuild, pending.value.request)
+			const { kept } = recomputable(manifest, rebuild, current, Date.now())
+			rebuildHours[rebuild.table] = [
+				...new Set([...(pending.value.rebuildHours[rebuild.table] ?? []), ...kept]),
+			].sort((a, b) => a - b)
+		}
+		yield* applyPending(db, manifest, dataDir, { ...pending.value, rebuildHours })
+		return Option.some(pending.value.request)
+	})
+
+/**
+ * Delete what `request` names. The caller holds exclusive access to the store
+ * (the server's admission gate), so no ingest or rollup interleaves.
+ */
+export const runScopedDelete = (
+	db: Db,
+	manifest: LocalSchemaManifest,
+	dataDir: string,
+	input: ScopedDeleteRequest,
+	options: { readonly dryRun: boolean; readonly nowMs?: number },
+): Effect.Effect<ScopedDeleteReport, ScopedDeleteError> =>
+	Effect.gen(function* () {
+		yield* assertPlanSound(manifest)
+		const nowMs = options.nowMs ?? Date.now()
+		const request: ScopedDeleteRequest =
+			input.beforeMs === undefined ? input : { ...input, beforeMs: floorToHour(input.beforeMs) }
+		if (!options.dryRun) yield* resumePendingDelete(db, manifest, dataDir)
+
+		const plan = resolvePlan(manifest, request)
+		const tables: ScopedDeleteTableReport[] = []
+		for (const { table, predicate } of plan.filters)
+			tables.push({ table, strategy: "filter", rows: yield* countRows(db, table, predicate) })
+		const rebuildHours: Record<string, number[]> = {}
+		for (const rebuild of plan.rebuilds) {
+			const hours = yield* affectedHours(db, rebuild, request)
+			const { kept, skipped } = recomputable(manifest, rebuild, hours, nowMs)
+			rebuildHours[rebuild.table] = kept
+			tables.push({
+				table: rebuild.table,
+				strategy: "rebuild",
+				rows:
+					kept.length === 0
+						? 0
+						: yield* countRows(db, rebuild.table, rebuildTargetPredicate(rebuild, request, kept)),
+				rebuiltHours: kept.length,
+				skippedHours: skipped.length,
+			})
+		}
+		for (const { table, reason } of plan.excluded) tables.push({ table, strategy: "excluded", rows: 0, reason })
+		tables.sort((a, b) => a.table.localeCompare(b.table))
+		const report: ScopedDeleteReport = {
+			request,
+			dryRun: options.dryRun,
+			deletedRows: tables.reduce((sum, t) => sum + (t.strategy === "filter" ? t.rows : 0), 0),
+			tables,
+		}
+		if (options.dryRun) return report
+
+		const pending: PendingDelete = { formatVersion: 1, request, rebuildHours }
+		yield* writeFile("write pending delete journal", () => durableJson(pendingDeletePath(dataDir), pending))
+		yield* applyPending(db, manifest, dataDir, pending)
+		return report
+	})
diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts
index 886036f93b..0d5244ff3f 100644
--- a/apps/cli/src/server/serve.ts
+++ b/apps/cli/src/server/serve.ts
@@ -38,7 +38,8 @@ import {
 	decodeTraceRequest,
 	encodeExportResponse,
 } from "./otlp/proto"
-import { CURRENT_LOCAL_SCHEMA, LOCAL_SCHEMA_SQL, SCHEMA_FINGERPRINT } from "./schema-identity"
+import { CURRENT_LOCAL_SCHEMA, LOCAL_SCHEMA_MANIFEST, LOCAL_SCHEMA_SQL, SCHEMA_FINGERPRINT } from "./schema-identity"
+import { resumePendingDelete, runScopedDelete, ScopedDeleteRequest } from "./scoped-delete"
 import { assertCurrentPhysicalSchema } from "./schema-physical"
 import {
 	countArrayRows,
@@ -610,6 +611,8 @@ interface ServerStatus {
 	url: string
 	readonly dataDir: string
 	lastIngestAtMs: number | null
+	/** Last scoped delete that may have changed the store; checkpoint refresh reads it. */
+	lastDeleteAtMs: number | null
 }
 
 /** OTLP-ingest request as a `Server`-kind span, mirroring the Rust gateway
@@ -943,6 +946,44 @@ const MAX_CHECKPOINT_BODY_BYTES = 4 * 1024
 const MAX_PROJECTION_BODY_BYTES = 512 * 1024
 const MAX_CONSUMER_BODY_BYTES = 16 * 1024
 const MAX_RETIREMENT_BODY_BYTES = 16 * 1024
+const MAX_DELETE_BODY_BYTES = 16 * 1024
+
+const ScopedDeleteBody = Schema.Struct({ ...ScopedDeleteRequest.fields, dryRun: Schema.Boolean })
+
+/** `maple delete`: removes one service's rows from raw and derived tables while admission is closed. */
+const handleScopedDelete = (
+	db: Chdb,
+	dataDir: string,
+	gate: RequestQuiescenceGate,
+	status: ServerStatus,
+	token: string,
+	req: Request,
+): Effect.Effect<Response> => {
+	if (!maintenanceTokenMatches(token, req.headers.get("x-maple-maintenance-token")))
+		return Effect.succeed(text("maintenance authorization required", 403))
+	return Effect.gen(function* () {
+		const body = yield* readBoundedJson(req, MAX_DELETE_BODY_BYTES)
+		const decoded = Schema.decodeUnknownResult(ScopedDeleteBody, { onExcessProperty: "error" })(body)
+		if (Result.isFailure(decoded)) return text("invalid delete fields", 400)
+		const { dryRun, ...request } = decoded.success
+		const work = runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, request, { dryRun })
+		const report = yield* gate.exclusiveEffect(
+			dryRun
+				? work
+				: // Even a failed delete may have changed rows, so the next checkpoint refresh must run.
+					work.pipe(Effect.ensuring(Effect.sync(() => (status.lastDeleteAtMs = Date.now())))),
+		)
+		return json(report)
+	}).pipe(
+		Effect.catchTags({
+			"@maple/cli/MaintenanceInProgress": (error) => Effect.succeed(text(error.message, 409)),
+			"@maple/cli/RequestBodyTooLarge": (error) => Effect.succeed(text(error.message, 413)),
+			"@maple/cli/InvalidJsonBody": () => Effect.succeed(text("invalid JSON body", 400)),
+			"@maple/cli/ScopedDeleteError": (error) =>
+				Effect.succeed(text(`scoped delete failed: ${error.message}`, 409)),
+		}),
+	)
+}
 
 /** The chDB half of a checkpoint backup failed. */
 class CheckpointBackupFailed extends Schema.TaggedError<CheckpointBackupFailed>()(
@@ -1315,6 +1356,7 @@ const statusResponse = (status: ServerStatus): Response =>
 		url: status.url,
 		dataDir: status.dataDir,
 		lastIngestAtMs: status.lastIngestAtMs,
+		lastDeleteAtMs: status.lastDeleteAtMs,
 	})
 
 /** The `Bun.serve` fetch handler, closed over the chDB connection. Each ingest
@@ -1377,6 +1419,12 @@ const makeFetch =
 				return respond(await runSpan(handleConsumerClaim(gate, consumerToken, req)))
 			if (url.pathname === "/local/eventing/acks")
 				return respond(await runSpan(handleConsumerAcknowledgement(gate, consumerToken, req)))
+			if (url.pathname === "/local/maintenance/delete")
+				return respond(
+					await runSpan(
+						handleScopedDelete(db, options.dataDir, gate, context.status, maintenanceToken, req),
+					),
+				)
 			if (url.pathname === "/local/retention/retire")
 				return respond(await handleRetirement(db, authority, gate, maintenanceToken, req))
 		}
@@ -1531,6 +1579,13 @@ export const startServer = (
 					message: `failed to enforce retired-day authority: ${describeThrown(error)}`,
 				}),
 		})
+		// A delete that failed or crashed midway is journaled; finish it before
+		// serving. A failure keeps the journal for the next `maple delete` to resume.
+		yield* resumePendingDelete(db, LOCAL_SCHEMA_MANIFEST, options.dataDir).pipe(
+			Effect.catchTag("@maple/cli/ScopedDeleteError", (error) =>
+				Effect.logWarning(`interrupted scoped delete is still pending: ${error.message}`),
+			),
+		)
 		const maintenanceToken = yield* Effect.tryPromise({
 			try: () => ensureMaintenanceToken(options.dataDir),
 			catch: (error) =>
@@ -1560,7 +1615,7 @@ export const startServer = (
 					),
 				),
 		)
-		const status: ServerStatus = { url: "", dataDir: resolve(options.dataDir), lastIngestAtMs: null }
+		const status: ServerStatus = { url: "", dataDir: resolve(options.dataDir), lastIngestAtMs: null, lastDeleteAtMs: null }
 		const server = yield* Effect.acquireRelease(
 			Effect.try({
 				try: () =>
@@ -1622,6 +1677,7 @@ export const __testables = {
 	handleConsumerRegistration,
 	handleCheckpointBackup,
 	handleEventingRead,
+	handleScopedDelete,
 	handleProjectionActivation,
 	ingest,
 	readBoundedJson,
diff --git a/apps/cli/test/delete-command.test.ts b/apps/cli/test/delete-command.test.ts
new file mode 100644
index 0000000000..adeb8b044c
--- /dev/null
+++ b/apps/cli/test/delete-command.test.ts
@@ -0,0 +1,53 @@
+import { afterEach, describe, expect, test } from "bun:test"
+import { Effect, Layer } from "effect"
+import { parseBefore, resolveDeleteBaseUrl } from "../src/commands/delete"
+import { Mode } from "../src/core/mode"
+import { CliUsageError } from "../src/lib/errors"
+
+const withMode = (resolved: Effect.Success<(typeof Mode.Service)["resolve"]>) =>
+	Effect.runPromise(
+		Effect.flip(resolveDeleteBaseUrl).pipe(
+			Effect.provide(Layer.succeed(Mode, { resolve: Effect.succeed(resolved) })),
+		),
+	)
+
+describe("maple delete is local-only", () => {
+	const argv = process.argv
+	afterEach(() => {
+		process.argv = argv
+	})
+
+	test("refuses --remote before resolving any backend", async () => {
+		process.argv = [...argv, "--remote"]
+		const error = await withMode({ _tag: "local", baseUrl: "http://127.0.0.1:4318" })
+		expect(error).toBeInstanceOf(CliUsageError)
+		expect(error.message).toContain("--remote is not supported")
+	})
+
+	test("refuses when the active mode resolves to remote", async () => {
+		const error = await withMode({ _tag: "remote", apiUrl: "https://api.maple.dev", token: "t", orgId: undefined })
+		expect(error).toBeInstanceOf(CliUsageError)
+		expect(error.message).toContain("the active mode is remote")
+	})
+
+	test("targets the resolved local server", async () => {
+		const url = await Effect.runPromise(
+			resolveDeleteBaseUrl.pipe(
+				Effect.provide(
+					Layer.succeed(Mode, {
+						resolve: Effect.succeed({ _tag: "local" as const, baseUrl: "http://127.0.0.1:4318/" }),
+					}),
+				),
+			),
+		)
+		expect(url).toBe("http://127.0.0.1:4318")
+	})
+})
+
+describe("--before", () => {
+	test("accepts an age or an absolute UTC timestamp", () => {
+		expect(parseBefore("2h", 10 * 3_600_000)).toBe(8 * 3_600_000)
+		expect(parseBefore("2026-10-01 12:00", 0)).toBe(Date.UTC(2026, 9, 1, 12))
+		expect(parseBefore("yesterday", 0)).toBeNull()
+	})
+})
diff --git a/apps/cli/test/scoped-delete.test.ts b/apps/cli/test/scoped-delete.test.ts
new file mode 100644
index 0000000000..20d8ca27a9
--- /dev/null
+++ b/apps/cli/test/scoped-delete.test.ts
@@ -0,0 +1,208 @@
+import { afterAll, beforeAll, describe, expect, test } from "bun:test"
+import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"
+import { homedir, tmpdir } from "node:os"
+import { join } from "node:path"
+import { Effect, Schema } from "effect"
+import schemaSql from "../src/server/schema/local-schema.sql" with { type: "text" }
+import { Chdb } from "../src/server/chdb"
+import { decodeJsonEachRow } from "../src/server/chdb-rows"
+import { buildLocalSchemaManifest } from "../src/server/schema-manifest"
+import { LOCAL_SCHEMA_MANIFEST } from "../src/server/schema-identity"
+import {
+	pendingDeletePath,
+	resumePendingDelete,
+	runScopedDelete,
+	TABLE_DELETE_PLAN,
+	tablePlan,
+	validateDeletePlan,
+} from "../src/server/scoped-delete"
+
+describe("scoped delete plan", () => {
+	// Fails when a table is added to the local schema without a delete strategy.
+	test("classifies every table in the bundled schema soundly", () => {
+		expect(validateDeletePlan(LOCAL_SCHEMA_MANIFEST)).toEqual([])
+		const tables = LOCAL_SCHEMA_MANIFEST.objects.filter((o) => o.kind === "table").map((o) => o.name)
+		expect(Object.keys(TABLE_DELETE_PLAN).sort()).toEqual([...tables].sort())
+	})
+
+	const withExtra = (ddl: string) => buildLocalSchemaManifest(`${schemaSql}\n\n${ddl}\n`)
+
+	test("refuses a new derived table nobody classified", () => {
+		const manifest = withExtra(
+			"CREATE TABLE IF NOT EXISTS traces_by_route (OrgId String, Hour DateTime, ServiceName String, Count UInt64)\nENGINE = SummingMergeTree\nORDER BY (OrgId, Hour);\n\nCREATE MATERIALIZED VIEW IF NOT EXISTS traces_by_route_mv TO traces_by_route AS\nSELECT OrgId, toStartOfHour(toDateTime(Timestamp)) AS Hour, ServiceName, count() AS Count FROM traces GROUP BY OrgId, Hour, ServiceName;",
+		)
+		expect(validateDeletePlan(manifest)).toContain("traces_by_route: no delete plan for this table")
+	})
+
+	test("refuses a filter on a merged column that is not a key", () => {
+		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
+			...TABLE_DELETE_PLAN,
+			error_fingerprints_minutely: { strategy: "filter", service: ["ServiceName"], env: null, time: "Minute" },
+		})
+		expect(problems.some((p) => p.includes("ServiceName is not in the sorting key"))).toBe(true)
+	})
+
+	test("refuses excluding a table that telemetry views write", () => {
+		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
+			...TABLE_DELETE_PLAN,
+			trace_list_mv: { strategy: "excluded", reason: "test" },
+		})
+		expect(problems).toContain("trace_list_mv_mv: writes excluded trace_list_mv from traces")
+	})
+})
+
+// Needs a real libchdb; skipped where none is installed.
+const libchdbAvailable =
+	(process.env.MAPLE_LIBCHDB !== undefined && existsSync(process.env.MAPLE_LIBCHDB)) ||
+	existsSync(join(homedir(), ".maple", "bin", "libchdb.so")) ||
+	existsSync(join(homedir(), ".maple", "bin", "libchdb.dylib"))
+
+const HOUR_MS = 3_600_000
+const iso = (ms: number) => new Date(ms).toISOString().replace("T", " ").replace("Z", "")
+const decodeCount = decodeJsonEachRow(Schema.Struct({ n: Schema.String }))
+
+describe.skipIf(!libchdbAvailable)("scoped delete against chDB", () => {
+	let root = ""
+	let dataDir = ""
+	let db: Chdb
+	const now = Date.now()
+	const hourAgo = Math.floor(now / HOUR_MS) * HOUR_MS - HOUR_MS + 60_000
+	const twoHoursAgo = hourAgo - HOUR_MS
+
+	const count = (sql: string) => Number(decodeCount(db.query(sql))[0]?.n)
+	const serviceRows = (table: string, service: string) => {
+		const plan = tablePlan(table)!
+		const columns = plan.strategy === "filter" ? plan.service : ["ServiceName"]
+		return count(
+			`SELECT toString(count()) AS n FROM ${table} WHERE ${columns.map((c) => `${c} = '${service}'`).join(" OR ")}`,
+		)
+	}
+	const attributeKey = (key: string) =>
+		count(`SELECT toString(sum(UsageCount)) AS n FROM attribute_keys_hourly WHERE AttributeKey = '${key}'`)
+
+	const span = (at: number, id: string, service: string, env: string, parent = "", kind = "Server") =>
+		`('local', '${iso(at)}', 't-${id}', 's-${id}', '${parent}', 'GET /x', '${kind}', '${service}', 1000000, 'Error', ` +
+		`map('deployment.environment', '${env}'), map('attr.${service}', 'v', 'shared.key', 'v', 'exception.type', 'E'), ` +
+		`[toDateTime64('${iso(at)}', 9)], ['exception'], [map('exception.type', 'Boom', 'exception.message', 'm')])`
+	const log = (at: number, service: string, env: string) =>
+		`('local', '${iso(at)}', '${iso(at).slice(0, 19)}', '${service}', 'body', map('deployment.environment', '${env}'), map('log.${service}', 'v'))`
+	const metric = (at: number, service: string, env: string) =>
+		`('local', map('deployment.environment', '${env}'), '${service}', 'm.${service}', map('k', 'v'), '${iso(at)}', '${iso(at)}', 1, 1)`
+
+	beforeAll(() => {
+		root = mkdtempSync(join(tmpdir(), "maple-scoped-delete-"))
+		dataDir = join(root, "data")
+		db = Chdb.open({ dataDir, schemaSql })
+		const spans = [
+			span(twoHoursAgo, "k1", "keep", "prod"),
+			span(hourAgo, "k2", "keep", "prod"),
+			span(twoHoursAgo, "d1", "drop", "prod"),
+			span(hourAgo, "d2", "drop", "pr-1"),
+			span(hourAgo + 1, "d3", "drop", "prod", "", "Client"),
+		]
+		db.exec(
+			`INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes, EventsTimestamp, EventsName, EventsAttributes) VALUES ${spans.join(", ")}`,
+		)
+		db.exec(
+			`INSERT INTO logs (OrgId, Timestamp, TimestampTime, ServiceName, Body, ResourceAttributes, LogAttributes) VALUES ${[
+				log(twoHoursAgo, "keep", "prod"),
+				log(hourAgo, "drop", "prod"),
+				log(hourAgo, "drop", "pr-1"),
+			].join(", ")}`,
+		)
+		db.exec(
+			`INSERT INTO metrics_sum (OrgId, ResourceAttributes, ServiceName, MetricName, Attributes, StartTimeUnix, TimeUnix, Value, IsMonotonic) VALUES ${[
+				metric(hourAgo, "keep", "prod"),
+				metric(hourAgo, "drop", "prod"),
+				metric(hourAgo, "drop", "pr-1"),
+			].join(", ")}`,
+		)
+		db.exec(
+			`INSERT INTO service_map_edges_hourly_ingest (OrgId, Hour, SourceService, TargetService, DeploymentEnv, CallCount) VALUES ('local', toStartOfHour(toDateTime('${iso(hourAgo).slice(0, 19)}')), 'keep', 'drop', 'prod', 3), ('local', toStartOfHour(toDateTime('${iso(hourAgo).slice(0, 19)}')), 'keep', 'db', 'prod', 2)`,
+		)
+	}, 120_000)
+
+	afterAll(() => {
+		db?.close()
+		if (root) rmSync(root, { recursive: true, force: true })
+	})
+
+	const deleteNow = (request: Parameters<typeof runScopedDelete>[3], dryRun = false) =>
+		Effect.runPromise(runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, request, { dryRun }))
+
+	test("seed reached the derived tables", () => {
+		for (const table of ["trace_list_mv", "service_overview_hourly", "error_events", "service_usage", "metric_catalog"])
+			expect(serviceRows(table, "drop")).toBeGreaterThan(0)
+		expect(attributeKey("attr.drop")).toBe(3)
+	})
+
+	test("dry run counts without deleting", async () => {
+		const report = await deleteNow({ service: "drop", env: "pr-1" }, true)
+		expect(report.dryRun).toBe(true)
+		expect(report.tables.find((t) => t.table === "traces")?.rows).toBe(1)
+		expect(serviceRows("traces", "drop")).toBe(3)
+		expect(existsSync(pendingDeletePath(dataDir))).toBe(false)
+	})
+
+	test("--env removes one environment and recomputes tables without one", async () => {
+		const usageBefore = count(
+			"SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'drop'",
+		)
+		expect(usageBefore).toBe(3)
+		const report = await deleteNow({ service: "drop", env: "pr-1" })
+		expect(report.tables.find((t) => t.table === "service_usage")?.strategy).toBe("rebuild")
+		expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'drop'")).toBe(2)
+		expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE DeploymentEnv = 'pr-1'")).toBe(0)
+		expect(
+			count("SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'drop'"),
+		).toBe(2)
+		expect(count("SELECT toString(sum(LogCount)) AS n FROM service_usage WHERE ServiceName = 'drop'")).toBe(1)
+		expect(attributeKey("attr.drop")).toBe(2)
+		expect(existsSync(pendingDeletePath(dataDir))).toBe(false)
+	})
+
+	test("--service removes every derived row and keeps other services intact", async () => {
+		const keepBefore = Object.fromEntries(
+			Object.entries(TABLE_DELETE_PLAN)
+				.filter(([, plan]) => plan.strategy === "filter")
+				.map(([table]) => [table, serviceRows(table, "keep")]),
+		)
+		const sharedBefore = attributeKey("shared.key")
+		await deleteNow({ service: "drop" })
+		for (const [table, plan] of Object.entries(TABLE_DELETE_PLAN)) {
+			if (plan.strategy !== "filter") continue
+			expect({ table, rows: serviceRows(table, "drop") }).toEqual({ table, rows: 0 })
+			// Edges from keep to drop go too: they were derived from drop's spans.
+			if (table !== "service_map_edges_hourly")
+				expect({ table, rows: serviceRows(table, "keep") }).toEqual({ table, rows: keepBefore[table]! })
+		}
+		expect(serviceRows("service_map_edges_hourly", "keep")).toBe(1)
+		expect(serviceRows("error_fingerprints_minutely", "drop")).toBe(0)
+		expect(attributeKey("attr.drop")).toBe(0)
+		expect(attributeKey("log.drop")).toBe(0)
+		expect(attributeKey("attr.keep")).toBe(2)
+		expect(attributeKey("shared.key")).toBe(sharedBefore - 2)
+	})
+
+	test("--before only removes hours before the floored cutoff", async () => {
+		const report = await deleteNow({ service: "keep", beforeMs: hourAgo + 30 * 60_000 })
+		expect(report.request.beforeMs).toBe(hourAgo - 60_000)
+		expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'keep'")).toBe(1)
+		expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE ServiceName = 'keep'")).toBe(1)
+		expect(attributeKey("attr.keep")).toBe(1)
+	})
+
+	test("a journaled delete resumes on the next run", async () => {
+		db.exec(
+			`INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes, EventsTimestamp, EventsName, EventsAttributes) VALUES ${span(hourAgo, "z1", "zombie", "prod")}`,
+		)
+		writeFileSync(
+			pendingDeletePath(dataDir),
+			JSON.stringify({ formatVersion: 1, request: { service: "zombie" }, rebuildHours: {} }),
+		)
+		await Effect.runPromise(resumePendingDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir))
+		expect(serviceRows("trace_list_mv", "zombie")).toBe(0)
+		expect(attributeKey("attr.zombie")).toBe(0)
+		expect(existsSync(pendingDeletePath(dataDir))).toBe(false)
+	})
+})
diff --git a/apps/cli/test/server-args.test.ts b/apps/cli/test/server-args.test.ts
index 85f167d45d..361c15c238 100644
--- a/apps/cli/test/server-args.test.ts
+++ b/apps/cli/test/server-args.test.ts
@@ -8,6 +8,7 @@ import {
 	defaultDataDir,
 	defaultLocalUrl,
 	ingestedSince,
+	storeChangedSince,
 	mapleCommand,
 	parseElapsedSeconds,
 	prettyPath,
@@ -241,6 +242,14 @@ describe("checkpoint refresh skip", () => {
 		strictEqual(ingestedSince(1_000, 1_000), true)
 		strictEqual(ingestedSince(undefined, 1_000), true) // status unknown: take it
 	})
+
+	it("also refreshes after a scoped delete, so restore points drop the deleted rows", () => {
+		const status = { service: "maple-local", pid: 1, version: "x", url: "", dataDir: "" } as const
+		strictEqual(storeChangedSince({ ...status, lastIngestAtMs: null }, 1_000), false)
+		strictEqual(storeChangedSince({ ...status, lastIngestAtMs: null, lastDeleteAtMs: 1_000 }, 1_000), true)
+		strictEqual(storeChangedSince({ ...status, lastIngestAtMs: 999, lastDeleteAtMs: 999 }, 1_000), false)
+		strictEqual(storeChangedSince(undefined, 1_000), true)
+	})
 })
 
 describe("prettyPath", () => {
diff --git a/docs/local-mode.md b/docs/local-mode.md
index 1cbacc3acb..ffae24a119 100644
--- a/docs/local-mode.md
+++ b/docs/local-mode.md
@@ -235,6 +235,83 @@ Checkpoints and `maple stop` wait for at most one hour's join. Sealed hours and
 retired UTC days are skipped. A span that arrives after its hour is sealed is
 not counted in the map's complete hours.
 
+### Deleting one service's data
+
+`maple delete` removes one service's telemetry from the live store without a
+full `maple reset`, for example when one shared local Maple serves many
+throwaway dev environments:
+
+```bash
+maple delete --service checkout-pr-42                 # preview: per-table counts, exits 1
+maple delete --service checkout-pr-42 --yes           # delete everything for the service
+maple delete --service api --env preview-17 --yes     # one deployment.environment only
+maple delete --service api --before 7d --yes          # rows older than an age or UTC timestamp
+```
+
+`--service` is required. `--env` matches `deployment.environment.name`, falling
+back to `deployment.environment`, the same expression the rollups store as
+`DeploymentEnv`. `--before` is floored to the UTC hour, because hourly rollups
+cannot split an hour; the report echoes the effective cutoff.
+
+The command is local-only: `--remote`, or a mode that resolves to remote, is
+refused before anything is sent. It needs a running `maple start`, reads the
+maintenance token beside the server's data directory (so only the machine and
+user owning the store can delete), and posts to `POST /local/maintenance/delete`.
+The read-only `/local/query` path is not involved. The server runs the delete
+inside the admission gate's exclusive section, so ingest, queries, the service
+map rollup and checkpoint backups wait for it.
+
+**Which tables.** Every table in the local schema is classified in
+`TABLE_DELETE_PLAN` (`apps/cli/src/server/scoped-delete.ts`), and
+`validateDeletePlan` checks the map against the bundled schema manifest. The
+server refuses to delete when they disagree, and
+`apps/cli/test/scoped-delete.test.ts` fails when a table is added without a
+classification. There are three strategies:
+
+- **filter**: `ALTER TABLE ... DELETE` on the service, env and time columns. Used
+  for the six raw tables (env from `ResourceAttributes`) and the derived tables
+  that key on `ServiceName` and `DeploymentEnv`: `trace_list_mv`, `error_events*`,
+  `service_overview_*`, `service_operations_*`, `traces_aggregates_hourly`,
+  `trace_facets_hourly`, the service map tables, and so on. On a merging engine
+  (Aggregating, Summing, Replacing) every filter column must be in the sorting
+  key, so a row never mixes services. `service_map_edges_hourly` and
+  `service_address_resolutions_hourly` match the service on either end of the
+  edge, because an edge is derived from both services' spans.
+- **rebuild**: tables whose rows do not carry the service as a key.
+  `attribute_keys_hourly` and `attribute_values_hourly` have no service column,
+  and `error_fingerprints_minutely` stores `ServiceName` as an `anyLast`
+  aggregate. Tables that key on the service but have no environment
+  (`service_usage`, `metric_catalog`, `span_metrics_calls_hourly`,
+  `product_events`, `ai_crawler_requests`) use this path only under `--env`.
+  The hours the deleted source rows covered are cleared and recomputed by
+  re-running the table's own materialized-view bodies over the surviving source
+  rows, one UTC day per insert. An hour is recomputed only while every source
+  still holds all of it (bundled TTL plus a day of margin). Older hours are left
+  as they are and reported as skipped; their TTL removes them.
+- **excluded**: tables no local telemetry reaches. `session_events`,
+  `session_replay_events`, `session_replays` and `identity_links` hold browser
+  session data that local OTLP ingest never writes. `alert_checks` and
+  `audit_log` are control-plane history, and `service_map_edges_hourly_ingest` is
+  a `Null` table that stores nothing. The validator also refuses any
+  materialized view that writes an excluded table from a deletable one.
+
+**Crash safety.** Before the first mutation the server durably writes
+`maple-pending-delete.json` beside the data directory, holding the request and
+the hours each rebuild will recompute. It is removed after the last step.
+Filter deletes are idempotent, so a delete that fails midway is finished at the
+next start (before the listener binds) or by the next `maple delete`.
+
+**Checkpoints and archives.** A delete changes only the live store.
+`/local/status` reports `lastDeleteAtMs`, and the checkpoint refresh treats a
+delete like ingest, so the next tick replaces `current`. Until two refreshes
+have rotated `previous` as well, and for pinned or reset-preserved checkpoints,
+`maple restore` brings the deleted rows back. Run `maple checkpoint` right after
+a delete to refresh `current` at once. Parquet archives are immutable exports
+of earlier checkpoints and are not touched; remove an archived day with
+`maple archive expire`. Re-archiving a day after a delete exports fewer rows,
+which `archive create` refuses without `--allow-shrink`. Retired days hold no
+live raw rows, so a delete never recomputes them.
+
 ### Versioned local-store migrations
 
 `maple start` never mutates a populated store in place when its schema identity
@@ -359,7 +436,7 @@ and every table function that reaches the filesystem or network are refused, and
 the statement runs with `readonly = 1` plus caps (30s, 4 GB memory, 1M rows,
 256 MiB result). A refusal is HTTP 400 with a body starting
 `read-only query endpoint: `. Internal writes (ingest, checkpoints, archives)
-use chDB directly; the native test probes write through `/local/query` with the
+and scoped deletes use chDB directly; the native test probes write through `/local/query` with the
 `x-maple-maintenance-token` header, a single-statement path reserved for them.
 
 The **server owns the output FORMAT**. `CH.compile(...)` appends
@@ -368,8 +445,9 @@ the statement as JSON rows, and keeps a client's own SETTINGS only where they
 lower the caps. Clients therefore POST `compiled.sql` verbatim.
 
 `GET /local/status` returns `{service: "maple-local", pid, version, url,
-dataDir, lastIngestAtMs}`, where `lastIngestAtMs` is when the server last
-accepted a non-empty OTLP batch of any signal. The CLI probe and the UI's
+dataDir, lastIngestAtMs, lastDeleteAtMs}`, where `lastIngestAtMs` is when the
+server last accepted a non-empty OTLP batch of any signal and `lastDeleteAtMs`
+is when a `maple delete` last ran (absent on older binaries). The CLI probe and the UI's
 connection pill use it; `/health` still answers `OK` for shell probes.
 
 Browser requests to `/local/*` are accepted only from the same origin, the
@@ -510,6 +588,7 @@ with the reason instead of returning a narrower answer:
 | `maple errors`                     | `/v2/error_issues` holds one issue per fingerprint, so it cannot report how many services an error spans.        |
 | `maple compare`                    | v2 has no window-comparison endpoint.                                                                            |
 | `maple diagnose`                   | Its error breakdown depends on the exception-type aggregates above.                                              |
+| `maple delete`                     | It deletes from the local store only (see [Deleting one service's data](#deleting-one-services-data)).           |
 
 `maple error <fp>` **does** work remotely: `/v2/error_issues?fingerprint_hash=`
 resolves the hash to an issue, and the issue detail carries the timeseries and
diff --git a/docs/local-telemetry-archives.md b/docs/local-telemetry-archives.md
index c49ea537e5..543f4e0c68 100644
--- a/docs/local-telemetry-archives.md
+++ b/docs/local-telemetry-archives.md
@@ -29,6 +29,9 @@ and the directory and manifest layouts.
   live store is a separate, explicit `maple archive retire-live` step.
 - Archive rehydration into the Maple UI. Historical data is queried in DuckDB,
   not reloaded into the dashboard.
+- Affected by `maple delete`. A scoped delete changes only the live store;
+  archived generations keep the deleted service's rows. Re-archiving an already
+  archived day after a delete needs `--allow-shrink`.
 - A second always-running database. Archives are files; DuckDB opens them on
   demand.
 

From 2cb56fb151c8fd27b96f689371e39406875960e9 Mon Sep 17 00:00:00 2001
From: Makisuo <minetoaster12@gmail.com>
Date: Fri, 2 Oct 2026 01:03:33 +0200
Subject: [PATCH 2/4] feat(cli): maple delete --namespace

Shared local Maples run throwaway environments as service.namespaces,
often with the same service names in each. `--namespace` deletes one
namespace's telemetry alone or together with `--service`.

Tables now declare which of service, namespace and env they filter
exactly; a request naming one a table lacks recomputes the affected
hours scoped to the dimensions it does key on, through its own views in
dependency order (so cascades like service_operations_minutely into
_hourly are wiped and redone) or, for the service map tables, by
re-running the hourly rollup for sealed hours. The validator checks the
plan for every flag combination.
---
 apps/cli/src/commands/delete.ts           |  48 +-
 apps/cli/src/server/scoped-delete.ts      | 650 ++++++++++++++--------
 apps/cli/src/server/serve.ts              |   5 +-
 apps/cli/src/server/service-map-rollup.ts |  31 +-
 apps/cli/test/delete-command.test.ts      |  20 +-
 apps/cli/test/scoped-delete.test.ts       | 119 +++-
 docs/local-mode.md                        |  74 +--
 docs/local-telemetry-archives.md          |   2 +-
 8 files changed, 649 insertions(+), 300 deletions(-)

diff --git a/apps/cli/src/commands/delete.ts b/apps/cli/src/commands/delete.ts
index 0c92999906..569547d8a4 100644
--- a/apps/cli/src/commands/delete.ts
+++ b/apps/cli/src/commands/delete.ts
@@ -8,7 +8,7 @@ import { parseTimestampMs, sinceToMs } from "../core/time"
 import { CliUsageError } from "../lib/errors"
 import { bold, dim, green } from "../lib/style"
 import { maintenanceTokenPath } from "../server/archives/retention"
-import { ScopedDeleteReport, type ScopedDeleteRequest } from "../server/scoped-delete"
+import { hasSubject, ScopedDeleteReport, ScopedDeleteRequest } from "../server/scoped-delete"
 import { jsonFormatRequested, writeJson } from "./json-output"
 import { LocalStatus, prettyPath } from "./server-args"
 
@@ -36,6 +36,22 @@ export const parseBefore = (raw: string, nowMs: number): number | null => {
 	return age === null ? parseTimestampMs(raw) : nowMs - age
 }
 
+/** Present flags as a request; a blank `--service` or `--namespace` counts as absent. */
+export const buildDeleteRequest = (flags: {
+	readonly service: string | undefined
+	readonly namespace: string | undefined
+	readonly env: string | undefined
+	readonly beforeMs: number | undefined
+}): Option.Option<ScopedDeleteRequest> => {
+	const present = Object.entries({
+		service: flags.service?.trim() || undefined,
+		namespace: flags.namespace?.trim() || undefined,
+		env: flags.env,
+		beforeMs: flags.beforeMs === undefined ? undefined : Math.floor(flags.beforeMs),
+	}).filter(([, value]) => value !== undefined)
+	return Option.filter(Schema.decodeUnknownOption(ScopedDeleteRequest)(Object.fromEntries(present)), hasSubject)
+}
+
 const failure = (message: string, hint?: string) =>
 	hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint })
 
@@ -77,7 +93,8 @@ const postDelete = (baseUrl: string, body: ScopedDeleteRequest & { readonly dryR
 
 const describeRequest = (request: ScopedDeleteRequest): string =>
 	[
-		`service ${bold(request.service)}`,
+		...(request.service === undefined ? [] : [`service ${bold(request.service)}`]),
+		...(request.namespace === undefined ? [] : [`namespace ${bold(request.namespace)}`]),
 		...(request.env === undefined ? [] : [`env ${bold(request.env === "" ? '""' : request.env)}`]),
 		...(request.beforeMs === undefined ? [] : [`before ${bold(new Date(request.beforeMs).toISOString())}`]),
 	].join(", ")
@@ -95,7 +112,14 @@ const renderReport = (report: ScopedDeleteReport): string => {
 }
 
 export const deleteCommand = Command.make("delete", {
-	service: Flag.String("service").pipe(Flag.withDescription("service.name whose telemetry to delete")),
+	service: Flag.optional(
+		Flag.String("service").pipe(Flag.withDescription("service.name whose telemetry to delete")),
+	),
+	namespace: Flag.optional(
+		Flag.String("namespace").pipe(
+			Flag.withDescription("service.namespace whose telemetry to delete (with --service: both must match)"),
+		),
+	),
 	env: Flag.optional(
 		Flag.String("env").pipe(
 			Flag.withDescription("Only rows whose deployment.environment(.name) equals this value"),
@@ -114,26 +138,30 @@ export const deleteCommand = Command.make("delete", {
 	),
 }).pipe(
 	Command.withDescription(
-		"Delete one service's telemetry from the local store: raw tables and every derived rollup (local only)",
+		"Delete a service's or namespace's telemetry from the local store: raw tables and every derived rollup (local only)",
 	),
 	Command.withExamples([
 		{ command: "maple delete --service checkout-pr-42" },
+		{ command: "maple delete --namespace pr-42 --yes" },
 		{ command: "maple delete --service api --env preview-17 --yes" },
 		{ command: "maple delete --service api --before 7d --yes" },
 	]),
 	Command.withHandler(
 		Effect.fnUntraced(function* (a) {
 			const baseUrl = yield* resolveDeleteBaseUrl
-			const service = a.service.trim()
-			if (service.length === 0) return yield* failure("--service must not be empty")
 			const rawBefore = Option.getOrUndefined(a.before)
 			const beforeMs = rawBefore === undefined ? undefined : parseBefore(rawBefore, Date.now())
 			if (beforeMs === null)
 				return yield* failure(`invalid --before: ${rawBefore}`, "use an age like 7d or a timestamp like 2026-10-01 12:00")
-			const env = Option.getOrUndefined(a.env)
-			const base = { service }
-			const withEnv = env === undefined ? base : { ...base, env }
-			const request = beforeMs === undefined ? withEnv : { ...withEnv, beforeMs: Math.floor(beforeMs) }
+			const built = buildDeleteRequest({
+				service: Option.getOrUndefined(a.service),
+				namespace: Option.getOrUndefined(a.namespace),
+				env: Option.getOrUndefined(a.env),
+				beforeMs,
+			})
+			if (Option.isNone(built))
+				return yield* failure("pass --service, --namespace, or both", "e.g. maple delete --namespace pr-42")
+			const request = built.value
 			const report = yield* postDelete(baseUrl, { ...request, dryRun: !a.yes })
 			if (jsonFormatRequested()) yield* writeJson(report)
 			else
diff --git a/apps/cli/src/server/scoped-delete.ts b/apps/cli/src/server/scoped-delete.ts
index c213d62291..1848a8c0a0 100644
--- a/apps/cli/src/server/scoped-delete.ts
+++ b/apps/cli/src/server/scoped-delete.ts
@@ -1,6 +1,6 @@
-// Scoped delete for the local store: one service's telemetry (optionally one
-// deployment.environment and/or everything before an hour) leaves the raw
-// tables and every table derived from them. See docs/local-mode.md.
+// Scoped delete for the local store: one service's or one service.namespace's
+// telemetry (optionally one deployment.environment and/or everything before an
+// hour) leaves the raw tables and every table derived from them. See docs/local-mode.md.
 
 import { Effect, Option, Schema } from "effect"
 import { existsSync } from "node:fs"
@@ -8,7 +8,13 @@ import type { Chdb } from "./chdb"
 import { decodeJsonEachRow, decodeRowCounts } from "./chdb-rows"
 import { durableJson, durableRemove } from "./durable-files"
 import { readRealFile } from "./local-token"
-import { type LocalSchemaManifest, ttlDaysFromDefinition, viewBody } from "./schema-manifest"
+import {
+	type LocalSchemaManifest,
+	type LocalSchemaObject,
+	ttlDaysFromDefinition,
+	viewBody,
+} from "./schema-manifest"
+import { serviceMapRollupInserts } from "./service-map-rollup"
 import { dataDirSidecarPath } from "./store-version"
 
 const HOUR_MS = 3_600_000
@@ -21,84 +27,94 @@ export class ScopedDeleteError extends Schema.TaggedError<ScopedDeleteError>()(
 	{ message: Schema.String, cause: Schema.optionalKey(Schema.Defect()) },
 ) {}
 
-/** Where a table's environment lives: a column, or the OTel resource map. */
-export type EnvSource = { readonly column: string } | { readonly resourceAttributes: string }
+/** The request dimensions a table can carry. */
+export type Dimension = "service" | "namespace" | "env"
+const DIMENSIONS: ReadonlyArray<Dimension> = ["service", "namespace", "env"]
 
-/** `filter` deletes by column (`env: null`: no env, so `--env` recomputes it), `rebuild`
- * recomputes affected hours from surviving source rows, `excluded` never holds local telemetry. */
+/** Where a dimension lives in a table: a column, or the OTel resource map. */
+export type DimensionSource = { readonly column: string } | { readonly resourceAttributes: string }
+
+/** `scoped` tables filter on the dimensions they declare and recompute when asked for one they lack. */
 export type TableDeletePlan =
 	| {
-			readonly strategy: "filter"
-			/** A row matches when any of these columns equals `--service`. */
-			readonly service: ReadonlyArray<string>
-			readonly env: EnvSource | null
+			readonly strategy: "scoped"
 			readonly time: string
+			/** Columns naming a service; a row matches when any equals `--service`. */
+			readonly service: ReadonlyArray<string> | null
+			readonly namespace: DimensionSource | null
+			readonly env: DimensionSource | null
+			/** Rebuild path when the request names an undeclared dimension. */
+			readonly recompute: "views" | "service-map-rollup" | null
 	  }
-	| { readonly strategy: "rebuild"; readonly time: string; readonly reason: string }
 	| { readonly strategy: "excluded"; readonly reason: string }
 
-const RESOURCE_ENV: EnvSource = { resourceAttributes: "ResourceAttributes" }
-const DEPLOYMENT_ENV: EnvSource = { column: "DeploymentEnv" }
+const RESOURCE: DimensionSource = { resourceAttributes: "ResourceAttributes" }
+const DEPLOYMENT_ENV: DimensionSource = { column: "DeploymentEnv" }
+const SERVICE_NAMESPACE: DimensionSource = { column: "ServiceNamespace" }
 
 const raw = (time: string): TableDeletePlan => ({
-	strategy: "filter",
-	service: ["ServiceName"],
-	env: RESOURCE_ENV,
+	strategy: "scoped",
 	time,
+	service: ["ServiceName"],
+	namespace: RESOURCE,
+	env: RESOURCE,
+	recompute: null,
 })
-const derived = (time: string, env: EnvSource | null = DEPLOYMENT_ENV): TableDeletePlan => ({
-	strategy: "filter",
+const derived = (
+	time: string,
+	dims: { readonly namespace?: DimensionSource; readonly env?: DimensionSource | null } = {},
+): TableDeletePlan => ({
+	strategy: "scoped",
+	time,
 	service: ["ServiceName"],
-	env,
+	namespace: dims.namespace ?? null,
+	env: dims.env === undefined ? DEPLOYMENT_ENV : dims.env,
+	recompute: "views",
+})
+/** Rows carry no usable service key (no column, or an aggregate), so they are always recomputed. */
+const unkeyed = (time: string): TableDeletePlan => ({
+	strategy: "scoped",
 	time,
+	service: null,
+	namespace: null,
+	env: null,
+	recompute: "views",
 })
+/** Edges come from both services' spans, so a row matches on either end. */
 const edges = (target: string): TableDeletePlan => ({
-	strategy: "filter",
+	strategy: "scoped",
+	time: "Hour",
 	service: ["SourceService", target],
+	namespace: null,
 	env: DEPLOYMENT_ENV,
-	time: "Hour",
+	recompute: "service-map-rollup",
 })
 const NOT_TELEMETRY = "control-plane history, not telemetry; local mode never writes it"
 const SESSION_DATA = "browser session data; local OTLP ingest never writes it"
 
-/**
- * Every table in the local schema. `validateDeletePlan` (and its test) fails when
- * the schema gains a table this map does not classify.
- */
 /** Delete strategy per table name; keys are checked against the schema at runtime. */
 export type TableDeletePlans = Readonly<Record<string, TableDeletePlan>>
 
+/** Every table in the local schema; `validateDeletePlan` fails when one is missing. */
 export const TABLE_DELETE_PLAN = {
-	ai_crawler_requests: derived("Timestamp", null),
+	ai_crawler_requests: derived("Timestamp", { env: null }),
 	ai_trace_index: derived("Timestamp"),
 	alert_checks: { strategy: "excluded", reason: NOT_TELEMETRY },
-	attribute_keys_hourly: {
-		strategy: "rebuild",
-		time: "Hour",
-		reason: "org-wide attribute discovery; rows have no service column",
-	},
-	attribute_values_hourly: {
-		strategy: "rebuild",
-		time: "Hour",
-		reason: "org-wide attribute discovery; rows have no service column",
-	},
+	attribute_keys_hourly: unkeyed("Hour"),
+	attribute_values_hourly: unkeyed("Hour"),
 	audit_log: { strategy: "excluded", reason: NOT_TELEMETRY },
 	error_events: derived("Timestamp"),
 	error_events_by_time: derived("Timestamp"),
-	error_fingerprints_minutely: {
-		strategy: "rebuild",
-		time: "Minute",
-		reason: "ServiceName is an anyLast aggregate, not a key; one row can mix services",
-	},
+	error_fingerprints_minutely: unkeyed("Minute"),
 	identity_links: { strategy: "excluded", reason: SESSION_DATA },
 	logs: raw("Timestamp"),
-	logs_aggregates_hourly: derived("Hour"),
-	metric_catalog: derived("Hour", null),
+	logs_aggregates_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }),
+	metric_catalog: derived("Hour", { env: null }),
 	metrics_exponential_histogram: raw("TimeUnix"),
 	metrics_gauge: raw("TimeUnix"),
 	metrics_histogram: raw("TimeUnix"),
 	metrics_sum: raw("TimeUnix"),
-	product_events: derived("Timestamp", null),
+	product_events: derived("Timestamp", { env: null }),
 	service_address_resolutions_hourly: edges("ResolvedTargetService"),
 	service_external_edges_hourly: derived("Hour"),
 	service_map_children: derived("Timestamp"),
@@ -112,18 +128,18 @@ export const TABLE_DELETE_PLAN = {
 	service_map_spans: derived("Timestamp"),
 	service_operations_hourly: derived("Hour"),
 	service_operations_minutely: derived("Minute"),
-	service_overview_hourly: derived("Hour"),
-	service_overview_minutely: derived("Minute"),
-	service_overview_spans: derived("Timestamp"),
+	service_overview_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }),
+	service_overview_minutely: derived("Minute", { namespace: SERVICE_NAMESPACE }),
+	service_overview_spans: derived("Timestamp", { namespace: SERVICE_NAMESPACE }),
 	service_platforms_hourly: derived("Hour"),
-	service_usage: derived("Hour", null),
+	service_usage: derived("Hour", { env: null }),
 	session_events: { strategy: "excluded", reason: SESSION_DATA },
 	session_replay_events: { strategy: "excluded", reason: SESSION_DATA },
 	session_replays: { strategy: "excluded", reason: SESSION_DATA },
-	span_metrics_calls_hourly: derived("Hour", null),
-	trace_detail_spans: derived("Timestamp", RESOURCE_ENV),
-	trace_facets_hourly: derived("Hour"),
-	trace_list_mv: derived("Timestamp"),
+	span_metrics_calls_hourly: derived("Hour", { env: null }),
+	trace_detail_spans: derived("Timestamp", { namespace: RESOURCE, env: RESOURCE }),
+	trace_facets_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }),
+	trace_list_mv: derived("Timestamp", { namespace: SERVICE_NAMESPACE }),
 	traces: raw("Timestamp"),
 	traces_aggregates_hourly: derived("Hour"),
 } satisfies TableDeletePlans
@@ -135,8 +151,73 @@ const lookup =
 /** The plan for any table name, or undefined when the schema has no such table. */
 export const tablePlan = lookup(TABLE_DELETE_PLAN)
 
+/** The service map rollup recomputes its two tables from raw spans only. */
+const ROLLUP_SOURCE = "traces"
+
+type ScopedPlan = Extract<TableDeletePlan, { strategy: "scoped" }>
+
+const declares = (entry: ScopedPlan, dimension: Dimension): boolean => entry[dimension] !== null
+
+// ---------------------------------------------------------------------------
+// Request and SQL.
+// ---------------------------------------------------------------------------
+
+export const ScopedDeleteRequest = Schema.Struct({
+	service: Schema.optionalKey(Schema.NonEmptyString.check(Schema.isMaxLength(512))),
+	namespace: Schema.optionalKey(Schema.NonEmptyString.check(Schema.isMaxLength(512))),
+	env: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(512))),
+	/** Exclusive cutoff, epoch ms; the server floors it to the UTC hour. */
+	beforeMs: Schema.optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))),
+})
+export type ScopedDeleteRequest = typeof ScopedDeleteRequest.Type
+
+/** A delete must name a service or a namespace; env and time only narrow it. */
+export const hasSubject = (request: ScopedDeleteRequest): boolean =>
+	request.service !== undefined || request.namespace !== undefined
+
+const requestedDimensions = (request: ScopedDeleteRequest): ReadonlyArray<Dimension> =>
+	DIMENSIONS.filter((dimension) => request[dimension] !== undefined)
+
+/** Bucketed rollups cannot split an hour, so every table uses the same hour cutoff. */
+export const floorToHour = (ms: number): number => Math.floor(ms / HOUR_MS) * HOUR_MS
+
+const sqlString = (value: string): string => `'${value.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'`
+
+const sourceExpression = (source: DimensionSource, dimension: Dimension): string => {
+	if ("column" in source) return source.column
+	const map = source.resourceAttributes
+	return dimension === "namespace"
+		? `${map}['service.namespace']`
+		: `coalesce(nullIf(${map}['deployment.environment.name'], ''), ${map}['deployment.environment'])`
+}
+
+/** `dimension = value` on this table; the caller has checked the table declares it. */
+const dimensionPredicate = (entry: ScopedPlan, dimension: Dimension, value: string): string => {
+	if (dimension === "service")
+		return `(${(entry.service ?? []).map((column) => `${column} = ${sqlString(value)}`).join(" OR ")})`
+	const source = entry[dimension]
+	return source === null ? "0" : `${sourceExpression(source, dimension)} = ${sqlString(value)}`
+}
+
+const scopePredicates = (entry: ScopedPlan, request: ScopedDeleteRequest, dims: ReadonlyArray<Dimension>) =>
+	dims.map((dimension) => dimensionPredicate(entry, dimension, request[dimension] ?? ""))
+
+const hourOf = (column: string) => `toUnixTimestamp(toStartOfHour(toDateTime(${column})))`
+
+/** The rows a request names, or null when the table lacks a requested dimension. */
+const filterPredicate = (entry: ScopedPlan, request: ScopedDeleteRequest): string | null => {
+	const dims = requestedDimensions(request)
+	if (!dims.every((dimension) => declares(entry, dimension))) return null
+	const parts = scopePredicates(entry, request, dims)
+	if (request.beforeMs !== undefined)
+		parts.push(`${entry.time} < toDateTime(${Math.floor(request.beforeMs / 1000)})`)
+	return parts.join(" AND ")
+}
+
+const hourList = (hours: ReadonlyArray<number>) => hours.join(", ")
+
 // ---------------------------------------------------------------------------
-// Schema graph.
+// Schema graph and plan resolution.
 // ---------------------------------------------------------------------------
 
 interface ViewEdge {
@@ -166,125 +247,182 @@ const materializedViews = (manifest: LocalSchemaManifest): ReadonlyArray<ViewEdg
 		})
 }
 
-const isRecomputable = (plan: TableDeletePlan | undefined): boolean =>
-	plan?.strategy === "rebuild" || (plan?.strategy === "filter" && plan.env === null)
+const fromPattern = (source: string) => new RegExp(`\\bFROM\\s+${source}\\b`, "gi")
+const sourceReferences = (body: string, source: string): number =>
+	[...body.matchAll(fromPattern(source))].length
+
+interface Rebuild {
+	readonly table: string
+	readonly time: string
+	readonly method: "views" | "service-map-rollup"
+	/** Requested dimensions this table keys on; its clear and re-insert are limited to them. */
+	readonly scope: ReadonlyArray<Dimension>
+	readonly views: ReadonlyArray<ViewEdge & { readonly source: string }>
+}
+
+interface ResolvedPlan {
+	readonly filters: ReadonlyArray<{ readonly table: string; readonly predicate: string }>
+	/** Topological: a rebuild never precedes the rebuild of a table it reads. */
+	readonly rebuilds: ReadonlyArray<Rebuild>
+	readonly excluded: ReadonlyArray<{ readonly table: string; readonly reason: string }>
+	/** Tables that can neither filter nor recompute this request. */
+	readonly unsupported: ReadonlyArray<string>
+}
+
+const resolvePlan = (
+	manifest: LocalSchemaManifest,
+	request: ScopedDeleteRequest,
+	plans: TableDeletePlans = TABLE_DELETE_PLAN,
+): ResolvedPlan => {
+	const views = materializedViews(manifest)
+	const requested = requestedDimensions(request)
+	const filters: Array<ResolvedPlan["filters"][number]> = []
+	const pending: Rebuild[] = []
+	const excluded: Array<ResolvedPlan["excluded"][number]> = []
+	const unsupported: string[] = []
+	for (const [table, entry] of Object.entries(plans)) {
+		if (entry.strategy === "excluded") {
+			excluded.push({ table, reason: entry.reason })
+			continue
+		}
+		const predicate = filterPredicate(entry, request)
+		if (predicate !== null) {
+			filters.push({ table, predicate })
+			continue
+		}
+		if (entry.recompute === null) {
+			unsupported.push(table)
+			continue
+		}
+		pending.push({
+			table,
+			time: entry.time,
+			method: entry.recompute,
+			// The rollup recomputes whole hours, so its clear is never narrowed.
+			scope:
+				entry.recompute === "views" ? requested.filter((dimension) => declares(entry, dimension)) : [],
+			views: views
+				.filter((view) => view.target === table)
+				.flatMap((view) => {
+					const source = view.sources[0]
+					return source !== undefined && plans[source] !== undefined && plans[source]!.strategy !== "excluded"
+						? [{ ...view, source }]
+						: []
+				}),
+		})
+	}
+	const rebuilds: Rebuild[] = []
+	while (pending.length > 0) {
+		const ready = pending.findIndex(
+			(rebuild) =>
+				!rebuild.views.some((view) => pending.some((other) => other !== rebuild && other.table === view.source)),
+		)
+		rebuilds.push(...pending.splice(ready === -1 ? 0 : ready, 1))
+	}
+	return { filters, rebuilds, excluded, unsupported }
+}
+
+/** Every flag combination the CLI can send, for validation. */
+const DELETE_SCOPES: ReadonlyArray<ScopedDeleteRequest> = [
+	{ service: "s" },
+	{ service: "s", env: "e" },
+	{ namespace: "n" },
+	{ namespace: "n", env: "e" },
+	{ service: "s", namespace: "n" },
+	{ service: "s", namespace: "n", env: "e" },
+]
+
+const scopeFlags = (request: ScopedDeleteRequest) =>
+	requestedDimensions(request)
+		.map((dimension) => `--${dimension}`)
+		.join(" ")
 
 /**
  * Everything that must hold for the plan to delete exactly the requested rows
- * from the given schema. Empty means sound; the server refuses otherwise.
+ * from the given schema, for every flag combination. Empty means sound.
  */
 export const validateDeletePlan = (
 	manifest: LocalSchemaManifest,
-	plan: TableDeletePlans = TABLE_DELETE_PLAN,
+	plans: TableDeletePlans = TABLE_DELETE_PLAN,
 ): ReadonlyArray<string> => {
 	const problems: string[] = []
 	const tables = new Map(manifest.objects.filter((o) => o.kind === "table").map((o) => [o.name, o]))
 	const views = materializedViews(manifest)
 	for (const name of tables.keys())
-		if (plan[name] === undefined) problems.push(`${name}: no delete plan for this table`)
-	for (const name of Object.keys(plan))
+		if (plans[name] === undefined) problems.push(`${name}: no delete plan for this table`)
+	for (const name of Object.keys(plans))
 		if (!tables.has(name)) problems.push(`${name}: planned but not in the schema`)
 
-	for (const [name, entry] of Object.entries(plan)) {
+	for (const [name, entry] of Object.entries(plans)) {
 		const table = tables.get(name)
-		if (table === undefined || entry.strategy === "excluded") continue
-		const columns = new Set(table.columns.map((c) => c.name))
-		const keyColumns = new Set(table.orderBy?.match(IDENTIFIER) ?? [])
-		// Only a plain MergeTree row is one event; any other engine merges rows by key.
-		const merges = table.engine !== "MergeTree"
-		const needColumn = (column: string, role: string) => {
-			if (!columns.has(column)) problems.push(`${name}: ${role} column ${column} does not exist`)
-			else if (merges && !keyColumns.has(column))
-				problems.push(`${name}: ${role} column ${column} is not in the sorting key of ${table.engine}`)
-		}
-		needColumn(entry.time, "time")
-		if (entry.strategy === "filter") {
-			for (const column of entry.service) needColumn(column, "service")
-			if (entry.env !== null) {
-				if ("column" in entry.env) needColumn(entry.env.column, "env")
-				else if (merges) problems.push(`${name}: a merging engine cannot filter env from a map`)
-				else if (!columns.has(entry.env.resourceAttributes))
-					problems.push(`${name}: env map ${entry.env.resourceAttributes} does not exist`)
+		if (table !== undefined && entry.strategy === "scoped") problems.push(...columnProblems(name, table, entry))
+	}
+
+	for (const request of DELETE_SCOPES) {
+		const label = scopeFlags(request)
+		const plan = resolvePlan(manifest, request, plans)
+		for (const table of plan.unsupported) problems.push(`${table}: cannot delete ${label} and has no recompute`)
+		const rebuilt = new Map(plan.rebuilds.map((rebuild) => [rebuild.table, rebuild]))
+		for (const rebuild of plan.rebuilds) {
+			if (rebuild.method === "service-map-rollup") continue
+			if (rebuild.views.length === 0) problems.push(`${rebuild.table}: recomputed for ${label} but no view writes it`)
+			const target = plans[rebuild.table]
+			if (rebuild.scope.includes("service") && target?.strategy === "scoped" && target.service?.join() !== "ServiceName")
+				problems.push(`${rebuild.table}: a service-scoped recompute must key on ServiceName alone`)
+			for (const view of rebuild.views) {
+				if (view.sources.length !== 1) problems.push(`${view.name}: recomputing needs exactly one source table`)
+				if (sourceReferences(view.body, view.source) !== 1)
+					problems.push(`${view.name}: recomputing needs exactly one FROM ${view.source}`)
+				const source = plans[view.source]
+				for (const dimension of rebuild.scope)
+					if (source?.strategy !== "scoped" || !declares(source, dimension))
+						problems.push(`${view.name}: source ${view.source} cannot scope ${dimension} for ${label}`)
 			}
-		}
-		if (!isRecomputable(entry)) continue
-		if (entry.strategy === "filter" && (entry.service.length !== 1 || entry.service[0] !== "ServiceName"))
-			problems.push(`${name}: a recomputed service-scoped table must key on ServiceName alone`)
-		const feeding = views.filter(
-			(view) => view.target === name && view.sources.some((s) => plan[s]?.strategy !== "excluded"),
-		)
-		if (feeding.length === 0) problems.push(`${name}: recomputed but no view writes it from telemetry`)
-		for (const view of feeding) {
-			if (view.sources.length !== 1) {
-				problems.push(`${view.name}: recomputing needs exactly one source table`)
-				continue
+			// Re-inserting fires the views this table feeds; their own recompute must wipe that.
+			for (const view of views.filter((candidate) => candidate.sources.includes(rebuild.table))) {
+				const downstream = rebuilt.get(view.target)
+				if (downstream === undefined)
+					problems.push(`${view.name}: recomputing ${rebuild.table} for ${label} leaks into ${view.target}`)
+				else if (!downstream.scope.every((dimension) => rebuild.scope.includes(dimension)))
+					problems.push(`${view.target}: recompute scope for ${label} is wider than ${rebuild.table}'s`)
 			}
-			const source = view.sources[0]!
-			const sourcePlan = plan[source]
-			if (sourcePlan?.strategy !== "filter" || sourcePlan.env === null)
-				problems.push(`${view.name}: source ${source} must be a filter table with an env dimension`)
-			if (sourceReferences(view.body, source) !== 1)
-				problems.push(`${view.name}: recomputing needs exactly one FROM ${source}`)
 		}
-		if (views.some((view) => view.sources.includes(name)))
-			problems.push(`${name}: recomputed rows would cascade into the views it feeds`)
 	}
 
 	// Rows derived from deletable telemetry must never land in an excluded table.
 	for (const view of views) {
-		if (plan[view.target]?.strategy !== "excluded") continue
+		if (plans[view.target]?.strategy !== "excluded") continue
 		for (const source of view.sources)
-			if (plan[source] !== undefined && plan[source]!.strategy !== "excluded")
+			if (plans[source] !== undefined && plans[source]!.strategy !== "excluded")
 				problems.push(`${view.name}: writes excluded ${view.target} from ${source}`)
 	}
-	return problems
+	return [...new Set(problems)]
 }
 
-const fromPattern = (source: string) => new RegExp(`\\bFROM\\s+${source}\\b`, "gi")
-const sourceReferences = (body: string, source: string): number =>
-	[...body.matchAll(fromPattern(source))].length
-
-// ---------------------------------------------------------------------------
-// Request and SQL.
-// ---------------------------------------------------------------------------
-
-export const ScopedDeleteRequest = Schema.Struct({
-	service: Schema.NonEmptyString.check(Schema.isMaxLength(512)),
-	env: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(512))),
-	/** Exclusive cutoff, epoch ms; the server floors it to the UTC hour. */
-	beforeMs: Schema.optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))),
-})
-export type ScopedDeleteRequest = typeof ScopedDeleteRequest.Type
-
-/** Bucketed rollups cannot split an hour, so every table uses the same hour cutoff. */
-export const floorToHour = (ms: number): number => Math.floor(ms / HOUR_MS) * HOUR_MS
-
-const sqlString = (value: string): string => `'${value.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'`
-
-const envExpression = (env: EnvSource): string =>
-	"column" in env
-		? env.column
-		: `coalesce(nullIf(${env.resourceAttributes}['deployment.environment.name'], ''), ${env.resourceAttributes}['deployment.environment'])`
-
-const hourOf = (column: string) => `toUnixTimestamp(toStartOfHour(toDateTime(${column})))`
-
-/** The rows a request names in a filter table, or null when `--env` cannot apply there. */
-const filterPredicate = (
-	entry: Extract<TableDeletePlan, { strategy: "filter" }>,
-	request: ScopedDeleteRequest,
-): string | null => {
-	if (request.env !== undefined && entry.env === null) return null
-	const service = sqlString(request.service)
-	const parts = [`(${entry.service.map((column) => `${column} = ${service}`).join(" OR ")})`]
-	if (request.env !== undefined && entry.env !== null)
-		parts.push(`${envExpression(entry.env)} = ${sqlString(request.env)}`)
-	if (request.beforeMs !== undefined)
-		parts.push(`${entry.time} < toDateTime(${Math.floor(request.beforeMs / 1000)})`)
-	return parts.join(" AND ")
+const columnProblems = (name: string, table: LocalSchemaObject, entry: ScopedPlan): ReadonlyArray<string> => {
+	const problems: string[] = []
+	const columns = new Set(table.columns.map((c) => c.name))
+	const keyColumns = new Set(table.orderBy?.match(IDENTIFIER) ?? [])
+	// Only a plain MergeTree row is one event; any other engine merges rows by key.
+	const merges = table.engine !== "MergeTree"
+	const needColumn = (column: string, role: string) => {
+		if (!columns.has(column)) problems.push(`${name}: ${role} column ${column} does not exist`)
+		else if (merges && !keyColumns.has(column))
+			problems.push(`${name}: ${role} column ${column} is not in the sorting key of ${table.engine}`)
+	}
+	needColumn(entry.time, "time")
+	for (const column of entry.service ?? []) needColumn(column, "service")
+	for (const dimension of ["namespace", "env"] as const) {
+		const source = entry[dimension]
+		if (source === null) continue
+		if ("column" in source) needColumn(source.column, dimension)
+		else if (merges) problems.push(`${name}: a merging engine cannot filter ${dimension} from a map`)
+		else if (!columns.has(source.resourceAttributes))
+			problems.push(`${name}: ${dimension} map ${source.resourceAttributes} does not exist`)
+	}
+	return problems
 }
 
-const hourList = (hours: ReadonlyArray<number>) => hours.join(", ")
-
 // ---------------------------------------------------------------------------
 // Execution.
 // ---------------------------------------------------------------------------
@@ -309,52 +447,6 @@ export const ScopedDeleteReport = Schema.Struct({
 })
 export type ScopedDeleteReport = typeof ScopedDeleteReport.Type
 
-interface Rebuild {
-	readonly table: string
-	readonly time: string
-	/** Set when only this service's rows are recomputed (a keyed table under `--env`). */
-	readonly serviceColumn: string | null
-	readonly views: ReadonlyArray<ViewEdge & { readonly source: string }>
-}
-
-interface ResolvedPlan {
-	readonly filters: ReadonlyArray<{ readonly table: string; readonly predicate: string }>
-	readonly rebuilds: ReadonlyArray<Rebuild>
-	readonly excluded: ReadonlyArray<{ readonly table: string; readonly reason: string }>
-}
-
-const resolvePlan = (manifest: LocalSchemaManifest, request: ScopedDeleteRequest): ResolvedPlan => {
-	const views = materializedViews(manifest)
-	const filters: Array<ResolvedPlan["filters"][number]> = []
-	const rebuilds: Rebuild[] = []
-	const excluded: Array<ResolvedPlan["excluded"][number]> = []
-	for (const [table, entry] of Object.entries(TABLE_DELETE_PLAN)) {
-		if (entry.strategy === "excluded") {
-			excluded.push({ table, reason: entry.reason })
-			continue
-		}
-		const predicate = entry.strategy === "filter" ? filterPredicate(entry, request) : null
-		if (predicate !== null) {
-			filters.push({ table, predicate })
-			continue
-		}
-		rebuilds.push({
-			table,
-			time: entry.time,
-			serviceColumn: entry.strategy === "filter" ? "ServiceName" : null,
-			views: views
-				.filter((view) => view.target === table)
-				.flatMap((view) => {
-					const source = view.sources[0]
-					return source !== undefined && tablePlan(source)?.strategy === "filter"
-						? [{ ...view, source }]
-						: []
-				}),
-		})
-	}
-	return { filters, rebuilds, excluded }
-}
-
 const run = <A>(label: string, f: () => A) =>
 	Effect.try({
 		try: f,
@@ -384,16 +476,39 @@ const distinctHours = (db: Db, table: string, time: string, predicate: string) =
 		).map((row) => Number(row.hour)),
 	)
 
-/** Hours (epoch seconds) a rebuild must recompute: the hours its sources lose rows in. */
-const affectedHours = (db: Db, rebuild: Rebuild, request: ScopedDeleteRequest) =>
+const scopedPlan = (table: string): Effect.Effect<ScopedPlan, ScopedDeleteError> => {
+	const entry = tablePlan(table)
+	return entry?.strategy === "scoped"
+		? Effect.succeed(entry)
+		: Effect.fail(new ScopedDeleteError({ message: `${table} has no scoped delete plan` }))
+}
+
+/** Hours a rebuild must recompute: where its sources lose rows, or are themselves recomputed. */
+const affectedHours = (
+	db: Db,
+	plan: ResolvedPlan,
+	rebuild: Rebuild,
+	request: ScopedDeleteRequest,
+	upstream: ReadonlyMap<string, ReadonlyArray<number>>,
+) =>
 	Effect.gen(function* () {
+		const sources =
+			rebuild.method === "service-map-rollup" ? [ROLLUP_SOURCE] : rebuild.views.map((view) => view.source)
 		const hours = new Set<number>()
-		for (const source of new Set(rebuild.views.map((view) => view.source))) {
-			const entry = tablePlan(source)
-			if (entry?.strategy !== "filter") continue
-			const predicate = filterPredicate(entry, request)
-			if (predicate === null) continue
-			for (const hour of yield* distinctHours(db, source, entry.time, predicate)) hours.add(hour)
+		for (const source of new Set(sources)) {
+			const filter = plan.filters.find((candidate) => candidate.table === source)
+			const found =
+				filter === undefined
+					? (upstream.get(source) ?? [])
+					: yield* distinctHours(db, source, (yield* scopedPlan(source)).time, filter.predicate)
+			for (const hour of found) hours.add(hour)
+		}
+		if (rebuild.method === "service-map-rollup" && hours.size > 0) {
+			// Only sealed hours: an open hour is still the rollup loop's to seal.
+			const sealed = new Set(
+				yield* distinctHours(db, "service_map_edges_hourly", "Hour", `${hourOf("Hour")} IN (${hourList([...hours])})`),
+			)
+			for (const hour of hours) if (!sealed.has(hour)) hours.delete(hour)
 		}
 		return [...hours].sort((a, b) => a - b)
 	})
@@ -409,10 +524,11 @@ const recomputable = (
 	hours: ReadonlyArray<number>,
 	nowMs: number,
 ) => {
-	const ttls = rebuild.views.map((view) => {
-		const definition = manifest.objects.find((o) => o.name === view.source)?.definition ?? ""
-		return ttlDaysFromDefinition(definition)
-	})
+	const sources =
+		rebuild.method === "service-map-rollup" ? [ROLLUP_SOURCE] : rebuild.views.map((view) => view.source)
+	const ttls = sources.map((source) =>
+		ttlDaysFromDefinition(manifest.objects.find((o) => o.name === source)?.definition ?? ""),
+	)
 	const kept: number[] = []
 	const skipped: number[] = []
 	for (const hour of hours) {
@@ -422,13 +538,24 @@ const recomputable = (
 	return { kept, skipped }
 }
 
-const rebuildTargetPredicate = (rebuild: Rebuild, request: ScopedDeleteRequest, hours: ReadonlyArray<number>) =>
-	[
-		...(rebuild.serviceColumn === null ? [] : [`${rebuild.serviceColumn} = ${sqlString(request.service)}`]),
-		`${hourOf(rebuild.time)} IN (${hourList(hours)})`,
-	].join(" AND ")
+const rebuildTargetPredicate = (
+	entry: ScopedPlan,
+	rebuild: Rebuild,
+	request: ScopedDeleteRequest,
+	hours: ReadonlyArray<number>,
+) => [...scopePredicates(entry, request, rebuild.scope), `${hourOf(rebuild.time)} IN (${hourList(hours)})`].join(" AND ")
+
+const clearHours = (db: Db, rebuild: Rebuild, request: ScopedDeleteRequest, hours: ReadonlyArray<number>) =>
+	Effect.gen(function* () {
+		const entry = yield* scopedPlan(rebuild.table)
+		yield* run(`clear ${rebuild.table}`, () =>
+			db.exec(
+				`ALTER TABLE ${rebuild.table} DELETE WHERE ${rebuildTargetPredicate(entry, rebuild, request, hours)} SETTINGS mutations_sync = 2`,
+			),
+		)
+	})
 
-const applyRebuild = (
+const recomputeViaViews = (
 	db: Db,
 	manifest: LocalSchemaManifest,
 	rebuild: Rebuild,
@@ -437,11 +564,7 @@ const applyRebuild = (
 ) =>
 	Effect.gen(function* () {
 		if (hours.length === 0) return
-		yield* run(`clear ${rebuild.table}`, () =>
-			db.exec(
-				`ALTER TABLE ${rebuild.table} DELETE WHERE ${rebuildTargetPredicate(rebuild, request, hours)} SETTINGS mutations_sync = 2`,
-			),
-		)
+		yield* clearHours(db, rebuild, request, hours)
 		// One UTC day per INSERT keeps each recompute's GROUP BY bounded.
 		const days = new Map<number, number[]>()
 		for (const hour of hours) {
@@ -453,8 +576,8 @@ const applyRebuild = (
 				.find((o) => o.name === view.source)
 				?.columns.map((c) => c.name)
 				.join(", ")
-			const sourceTime = tablePlan(view.source)
-			if (sourceColumns === undefined || sourceTime?.strategy !== "filter")
+			const source = yield* scopedPlan(view.source)
+			if (sourceColumns === undefined)
 				return yield* new ScopedDeleteError({ message: `${view.name}: unknown source ${view.source}` })
 			// The view's own output columns, so the INSERT matches by name, not position.
 			const columns = yield* run(`columns of ${view.name}`, () =>
@@ -468,8 +591,8 @@ const applyRebuild = (
 				return yield* new ScopedDeleteError({ message: `${view.name}: no columns in system.columns` })
 			for (const dayHours of days.values()) {
 				const scope = [
-					...(rebuild.serviceColumn === null ? [] : [`ServiceName = ${sqlString(request.service)}`]),
-					`${hourOf(sourceTime.time)} IN (${hourList(dayHours)})`,
+					...scopePredicates(source, request, rebuild.scope),
+					`${hourOf(source.time)} IN (${hourList(dayHours)})`,
 				].join(" AND ")
 				const body = view.body.replace(
 					fromPattern(view.source),
@@ -483,6 +606,29 @@ const applyRebuild = (
 		}
 	})
 
+/** Clear whole hours of the rollup's tables, then seal each hour again from surviving spans. */
+const recomputeServiceMap = (
+	db: Db,
+	rebuilds: ReadonlyArray<Rebuild>,
+	request: ScopedDeleteRequest,
+	hoursByTable: Readonly<Record<string, ReadonlyArray<number>>>,
+) =>
+	Effect.gen(function* () {
+		const hours = [...new Set(rebuilds.flatMap((rebuild) => hoursByTable[rebuild.table] ?? []))].sort(
+			(a, b) => a - b,
+		)
+		if (hours.length === 0) return
+		for (const rebuild of rebuilds) yield* clearHours(db, rebuild, request, hours)
+		for (const hour of hours) {
+			const statements = yield* serviceMapRollupInserts(hour * 1000).pipe(
+				Effect.mapError(
+					(cause) => new ScopedDeleteError({ message: `service map rollup: ${cause.message}`, cause }),
+				),
+			)
+			for (const statement of statements) yield* run("recompute service map", () => db.exec(statement))
+		}
+	})
+
 const applyFilters = (db: Db, filters: ResolvedPlan["filters"]) =>
 	Effect.forEach(
 		filters,
@@ -542,13 +688,48 @@ const assertPlanSound = (manifest: LocalSchemaManifest) => {
 			)
 }
 
+const assertSubject = (request: ScopedDeleteRequest) =>
+	hasSubject(request)
+		? Effect.void
+		: Effect.fail(new ScopedDeleteError({ message: "a scoped delete needs a service or a namespace" }))
+
+/** Hours per rebuild, in plan order so recomputed sources feed their dependants. */
+const planHours = (
+	db: Db,
+	manifest: LocalSchemaManifest,
+	plan: ResolvedPlan,
+	request: ScopedDeleteRequest,
+	nowMs: number,
+) =>
+	Effect.gen(function* () {
+		const all = new Map<string, ReadonlyArray<number>>()
+		const kept: Record<string, number[]> = {}
+		const skipped: Record<string, number> = {}
+		for (const rebuild of plan.rebuilds) {
+			const hours = yield* affectedHours(db, plan, rebuild, request, all)
+			all.set(rebuild.table, hours)
+			const split = recomputable(manifest, rebuild, hours, nowMs)
+			kept[rebuild.table] = split.kept
+			skipped[rebuild.table] = split.skipped.length
+		}
+		return { kept, skipped }
+	})
+
 /** Apply a journaled request: filters are idempotent, rebuilds take their hours from the journal. */
 const applyPending = (db: Db, manifest: LocalSchemaManifest, dataDir: string, pending: PendingDelete) =>
 	Effect.gen(function* () {
 		const plan = resolvePlan(manifest, pending.request)
 		yield* applyFilters(db, plan.filters)
+		// Views in dependency order, so each recompute wipes what its source's re-insert cascaded.
 		for (const rebuild of plan.rebuilds)
-			yield* applyRebuild(db, manifest, rebuild, pending.request, pending.rebuildHours[rebuild.table] ?? [])
+			if (rebuild.method === "views")
+				yield* recomputeViaViews(db, manifest, rebuild, pending.request, pending.rebuildHours[rebuild.table] ?? [])
+		yield* recomputeServiceMap(
+			db,
+			plan.rebuilds.filter((rebuild) => rebuild.method === "service-map-rollup"),
+			pending.request,
+			pending.rebuildHours,
+		)
 		yield* writeFile("remove pending delete journal", () => durableRemove(pendingDeletePath(dataDir)))
 	})
 
@@ -558,15 +739,14 @@ export const resumePendingDelete = (db: Db, manifest: LocalSchemaManifest, dataD
 		const pending = yield* readPending(dataDir)
 		if (Option.isNone(pending)) return Option.none<ScopedDeleteRequest>()
 		yield* assertPlanSound(manifest)
+		yield* assertSubject(pending.value.request)
 		// A restored checkpoint can bring the rows back; recompute their hours too.
+		const current = yield* planHours(db, manifest, resolvePlan(manifest, pending.value.request), pending.value.request, Date.now())
 		const rebuildHours: Record<string, number[]> = {}
-		for (const rebuild of resolvePlan(manifest, pending.value.request).rebuilds) {
-			const current = yield* affectedHours(db, rebuild, pending.value.request)
-			const { kept } = recomputable(manifest, rebuild, current, Date.now())
-			rebuildHours[rebuild.table] = [
-				...new Set([...(pending.value.rebuildHours[rebuild.table] ?? []), ...kept]),
+		for (const table of new Set([...Object.keys(pending.value.rebuildHours), ...Object.keys(current.kept)]))
+			rebuildHours[table] = [
+				...new Set([...(pending.value.rebuildHours[table] ?? []), ...(current.kept[table] ?? [])]),
 			].sort((a, b) => a - b)
-		}
 		yield* applyPending(db, manifest, dataDir, { ...pending.value, rebuildHours })
 		return Option.some(pending.value.request)
 	})
@@ -584,29 +764,33 @@ export const runScopedDelete = (
 ): Effect.Effect<ScopedDeleteReport, ScopedDeleteError> =>
 	Effect.gen(function* () {
 		yield* assertPlanSound(manifest)
+		yield* assertSubject(input)
 		const nowMs = options.nowMs ?? Date.now()
 		const request: ScopedDeleteRequest =
 			input.beforeMs === undefined ? input : { ...input, beforeMs: floorToHour(input.beforeMs) }
 		if (!options.dryRun) yield* resumePendingDelete(db, manifest, dataDir)
 
 		const plan = resolvePlan(manifest, request)
+		if (plan.unsupported.length > 0)
+			return yield* new ScopedDeleteError({
+				message: `cannot delete from ${plan.unsupported.join(", ")} for this request`,
+			})
 		const tables: ScopedDeleteTableReport[] = []
 		for (const { table, predicate } of plan.filters)
 			tables.push({ table, strategy: "filter", rows: yield* countRows(db, table, predicate) })
-		const rebuildHours: Record<string, number[]> = {}
+		const hours = yield* planHours(db, manifest, plan, request, nowMs)
 		for (const rebuild of plan.rebuilds) {
-			const hours = yield* affectedHours(db, rebuild, request)
-			const { kept, skipped } = recomputable(manifest, rebuild, hours, nowMs)
-			rebuildHours[rebuild.table] = kept
+			const kept = hours.kept[rebuild.table] ?? []
+			const entry = yield* scopedPlan(rebuild.table)
 			tables.push({
 				table: rebuild.table,
 				strategy: "rebuild",
 				rows:
 					kept.length === 0
 						? 0
-						: yield* countRows(db, rebuild.table, rebuildTargetPredicate(rebuild, request, kept)),
+						: yield* countRows(db, rebuild.table, rebuildTargetPredicate(entry, rebuild, request, kept)),
 				rebuiltHours: kept.length,
-				skippedHours: skipped.length,
+				skippedHours: hours.skipped[rebuild.table] ?? 0,
 			})
 		}
 		for (const { table, reason } of plan.excluded) tables.push({ table, strategy: "excluded", rows: 0, reason })
@@ -619,7 +803,7 @@ export const runScopedDelete = (
 		}
 		if (options.dryRun) return report
 
-		const pending: PendingDelete = { formatVersion: 1, request, rebuildHours }
+		const pending: PendingDelete = { formatVersion: 1, request, rebuildHours: hours.kept }
 		yield* writeFile("write pending delete journal", () => durableJson(pendingDeletePath(dataDir), pending))
 		yield* applyPending(db, manifest, dataDir, pending)
 		return report
diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts
index 0d5244ff3f..6cf5f98791 100644
--- a/apps/cli/src/server/serve.ts
+++ b/apps/cli/src/server/serve.ts
@@ -39,7 +39,7 @@ import {
 	encodeExportResponse,
 } from "./otlp/proto"
 import { CURRENT_LOCAL_SCHEMA, LOCAL_SCHEMA_MANIFEST, LOCAL_SCHEMA_SQL, SCHEMA_FINGERPRINT } from "./schema-identity"
-import { resumePendingDelete, runScopedDelete, ScopedDeleteRequest } from "./scoped-delete"
+import { hasSubject, resumePendingDelete, runScopedDelete, ScopedDeleteRequest } from "./scoped-delete"
 import { assertCurrentPhysicalSchema } from "./schema-physical"
 import {
 	countArrayRows,
@@ -950,7 +950,7 @@ const MAX_DELETE_BODY_BYTES = 16 * 1024
 
 const ScopedDeleteBody = Schema.Struct({ ...ScopedDeleteRequest.fields, dryRun: Schema.Boolean })
 
-/** `maple delete`: removes one service's rows from raw and derived tables while admission is closed. */
+/** `maple delete`: removes a service's or namespace's rows from raw and derived tables while admission is closed. */
 const handleScopedDelete = (
 	db: Chdb,
 	dataDir: string,
@@ -966,6 +966,7 @@ const handleScopedDelete = (
 		const decoded = Schema.decodeUnknownResult(ScopedDeleteBody, { onExcessProperty: "error" })(body)
 		if (Result.isFailure(decoded)) return text("invalid delete fields", 400)
 		const { dryRun, ...request } = decoded.success
+		if (!hasSubject(request)) return text("a delete needs service or namespace", 400)
 		const work = runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, request, { dryRun })
 		const report = yield* gate.exclusiveEffect(
 			dryRun
diff --git a/apps/cli/src/server/service-map-rollup.ts b/apps/cli/src/server/service-map-rollup.ts
index 43b970fa37..258ad1fa00 100644
--- a/apps/cli/src/server/service-map-rollup.ts
+++ b/apps/cli/src/server/service-map-rollup.ts
@@ -139,6 +139,18 @@ const sealedHours = (db: LocalServiceMapRollupDeps["db"], oldestHourMs: number,
 		return CH.serviceMapHourSet(yield* decodeRows(decodeHourRows, text, "existing hours"))
 	})
 
+/** The INSERTs that seal one hour from raw spans: resolutions, then edges (the seal). */
+export const serviceMapRollupInserts = (hourMs: number) =>
+	Effect.gen(function* () {
+		const params = CH.serviceMapRollupHourParams(ORG_ID, hourMs)
+		const resolutions = yield* selectBody(CH.serviceMapResolutionsRollupSQL(params), "resolutions")
+		const edges = yield* selectBody(CH.serviceMapEdgesRollupSQL(params), "edges")
+		return [
+			`INSERT INTO service_address_resolutions_hourly (${RESOLUTION_COLUMNS}) SELECT ${RESOLUTION_COLUMNS} FROM (${resolutions})`,
+			`INSERT INTO service_map_edges_hourly_ingest (${EDGE_COLUMNS}) SELECT ${EDGE_COLUMNS} FROM (${edges})`,
+		] as const
+	})
+
 /**
  * Roll up one hour inside the admission gate. Resolutions are written first and
  * the edges (the seal) last, so a failure leaves the hour unsealed and the next
@@ -151,22 +163,9 @@ const rollupHour = (deps: LocalServiceMapRollupDeps, hourMs: number) =>
 			leave === null
 				? Effect.succeed(false)
 				: Effect.gen(function* () {
-						const params = CH.serviceMapRollupHourParams(ORG_ID, hourMs)
-						const resolutions = yield* selectBody(
-							CH.serviceMapResolutionsRollupSQL(params),
-							"resolutions",
-						)
-						const edges = yield* selectBody(CH.serviceMapEdgesRollupSQL(params), "edges")
-						yield* runExec(
-							deps.db,
-							`INSERT INTO service_address_resolutions_hourly (${RESOLUTION_COLUMNS}) SELECT ${RESOLUTION_COLUMNS} FROM (${resolutions})`,
-							"insert address resolutions",
-						)
-						yield* runExec(
-							deps.db,
-							`INSERT INTO service_map_edges_hourly_ingest (${EDGE_COLUMNS}) SELECT ${EDGE_COLUMNS} FROM (${edges})`,
-							"insert service map edges",
-						)
+						const [resolutions, edges] = yield* serviceMapRollupInserts(hourMs)
+						yield* runExec(deps.db, resolutions, "insert address resolutions")
+						yield* runExec(deps.db, edges, "insert service map edges")
 						return true
 					}),
 		(leave) => Effect.sync(() => leave?.()),
diff --git a/apps/cli/test/delete-command.test.ts b/apps/cli/test/delete-command.test.ts
index adeb8b044c..2777116b89 100644
--- a/apps/cli/test/delete-command.test.ts
+++ b/apps/cli/test/delete-command.test.ts
@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, test } from "bun:test"
-import { Effect, Layer } from "effect"
-import { parseBefore, resolveDeleteBaseUrl } from "../src/commands/delete"
+import { Effect, Layer, Option } from "effect"
+import { buildDeleteRequest, parseBefore, resolveDeleteBaseUrl } from "../src/commands/delete"
 import { Mode } from "../src/core/mode"
 import { CliUsageError } from "../src/lib/errors"
 
@@ -51,3 +51,19 @@ describe("--before", () => {
 		expect(parseBefore("yesterday", 0)).toBeNull()
 	})
 })
+
+describe("delete request", () => {
+	const flags = { service: undefined, namespace: undefined, env: undefined, beforeMs: undefined }
+
+	test("needs a service or a namespace", () => {
+		expect(Option.isNone(buildDeleteRequest(flags))).toBe(true)
+		expect(Option.isNone(buildDeleteRequest({ ...flags, service: "  ", env: "prod" }))).toBe(true)
+	})
+
+	test("accepts a namespace alone or with a service, env and cutoff", () => {
+		expect(buildDeleteRequest({ ...flags, namespace: "pr-42" })).toEqual(Option.some({ namespace: "pr-42" }))
+		expect(
+			buildDeleteRequest({ service: " api ", namespace: "pr-42", env: "", beforeMs: 1_000.7 }),
+		).toEqual(Option.some({ service: "api", namespace: "pr-42", env: "", beforeMs: 1_000 }))
+	})
+})
diff --git a/apps/cli/test/scoped-delete.test.ts b/apps/cli/test/scoped-delete.test.ts
index 20d8ca27a9..473ff12545 100644
--- a/apps/cli/test/scoped-delete.test.ts
+++ b/apps/cli/test/scoped-delete.test.ts
@@ -8,6 +8,7 @@ import { Chdb } from "../src/server/chdb"
 import { decodeJsonEachRow } from "../src/server/chdb-rows"
 import { buildLocalSchemaManifest } from "../src/server/schema-manifest"
 import { LOCAL_SCHEMA_MANIFEST } from "../src/server/schema-identity"
+import { serviceMapRollupInserts } from "../src/server/service-map-rollup"
 import {
 	pendingDeletePath,
 	resumePendingDelete,
@@ -37,11 +38,48 @@ describe("scoped delete plan", () => {
 	test("refuses a filter on a merged column that is not a key", () => {
 		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
 			...TABLE_DELETE_PLAN,
-			error_fingerprints_minutely: { strategy: "filter", service: ["ServiceName"], env: null, time: "Minute" },
+			error_fingerprints_minutely: {
+				strategy: "scoped",
+				time: "Minute",
+				service: ["ServiceName"],
+				namespace: null,
+				env: null,
+				recompute: "views",
+			},
 		})
 		expect(problems.some((p) => p.includes("ServiceName is not in the sorting key"))).toBe(true)
 	})
 
+	test("refuses a table that can neither filter nor recompute a namespace", () => {
+		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
+			...TABLE_DELETE_PLAN,
+			service_map_spans: {
+				strategy: "scoped",
+				time: "Timestamp",
+				service: ["ServiceName"],
+				namespace: null,
+				env: { column: "DeploymentEnv" },
+				recompute: null,
+			},
+		})
+		expect(problems).toContain("service_map_spans: cannot delete --namespace and has no recompute")
+	})
+
+	test("refuses recomputing a view source whose dependant is filtered", () => {
+		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
+			...TABLE_DELETE_PLAN,
+			service_operations_hourly: {
+				strategy: "scoped",
+				time: "Hour",
+				service: ["ServiceName"],
+				namespace: { resourceAttributes: "ResourceAttributes" },
+				env: { column: "DeploymentEnv" },
+				recompute: "views",
+			},
+		})
+		expect(problems.some((p) => p.includes("leaks into service_operations_hourly"))).toBe(true)
+	})
+
 	test("refuses excluding a table that telemetry views write", () => {
 		const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, {
 			...TABLE_DELETE_PLAN,
@@ -72,7 +110,7 @@ describe.skipIf(!libchdbAvailable)("scoped delete against chDB", () => {
 	const count = (sql: string) => Number(decodeCount(db.query(sql))[0]?.n)
 	const serviceRows = (table: string, service: string) => {
 		const plan = tablePlan(table)!
-		const columns = plan.strategy === "filter" ? plan.service : ["ServiceName"]
+		const columns = plan.strategy === "scoped" && plan.service !== null ? plan.service : ["ServiceName"]
 		return count(
 			`SELECT toString(count()) AS n FROM ${table} WHERE ${columns.map((c) => `${c} = '${service}'`).join(" OR ")}`,
 		)
@@ -164,13 +202,13 @@ describe.skipIf(!libchdbAvailable)("scoped delete against chDB", () => {
 	test("--service removes every derived row and keeps other services intact", async () => {
 		const keepBefore = Object.fromEntries(
 			Object.entries(TABLE_DELETE_PLAN)
-				.filter(([, plan]) => plan.strategy === "filter")
+				.filter(([, plan]) => plan.strategy === "scoped" && plan.service !== null)
 				.map(([table]) => [table, serviceRows(table, "keep")]),
 		)
 		const sharedBefore = attributeKey("shared.key")
 		await deleteNow({ service: "drop" })
 		for (const [table, plan] of Object.entries(TABLE_DELETE_PLAN)) {
-			if (plan.strategy !== "filter") continue
+			if (plan.strategy !== "scoped" || plan.service === null) continue
 			expect({ table, rows: serviceRows(table, "drop") }).toEqual({ table, rows: 0 })
 			// Edges from keep to drop go too: they were derived from drop's spans.
 			if (table !== "service_map_edges_hourly")
@@ -206,3 +244,76 @@ describe.skipIf(!libchdbAvailable)("scoped delete against chDB", () => {
 		expect(existsSync(pendingDeletePath(dataDir))).toBe(false)
 	})
 })
+
+describe.skipIf(!libchdbAvailable)("namespace delete against chDB", () => {
+	let root = ""
+	let dataDir = ""
+	let db: Chdb
+	const hourMs = Math.floor(Date.now() / HOUR_MS) * HOUR_MS - 2 * HOUR_MS
+	const at = hourMs + 10 * 60_000
+
+	const count = (sql: string) => Number(decodeCount(db.query(sql))[0]?.n)
+	// A client span calling a child server span named "api", both in `namespace`.
+	const call = (id: string, client: string, namespace: string) => {
+		const resource = (service: string) =>
+			`map('service.name', '${service}', 'service.namespace', '${namespace}', 'deployment.environment', 'prod')`
+		return (
+			`('local', '${iso(at)}', 't-${id}', 'c-${id}', '', 'GET /api', 'Client', '${client}', 5000000, 'Ok', ${resource(client)}, map('server.address', 'api.internal')), ` +
+			`('local', '${iso(at + 1)}', 't-${id}', 's-${id}', 'c-${id}', 'GET /api', 'Server', 'api', 4000000, 'Error', ${resource("api")}, map('exception.type', 'E'))`
+		)
+	}
+	const edge = (source: string) =>
+		count(
+			`SELECT toString(sum(CallCount)) AS n FROM service_map_edges_hourly WHERE SourceService = '${source}' AND TargetService = 'api'`,
+		)
+
+	beforeAll(async () => {
+		root = mkdtempSync(join(tmpdir(), "maple-scoped-delete-ns-"))
+		dataDir = join(root, "data")
+		db = Chdb.open({ dataDir, schemaSql })
+		db.exec(
+			`INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes) VALUES ${call("a", "web", "pr-1")}, ${call("b", "caller", "pr-2")}`,
+		)
+		for (const statement of await Effect.runPromise(serviceMapRollupInserts(hourMs))) db.exec(statement)
+	}, 120_000)
+
+	afterAll(() => {
+		db?.close()
+		if (root) rmSync(root, { recursive: true, force: true })
+	})
+
+	test("removes one namespace and keeps a same-named service in another", async () => {
+		expect(edge("web")).toBe(1)
+		expect(edge("caller")).toBe(1)
+		expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_hourly WHERE ServiceName = 'api'")).toBe(2)
+
+		const report = await Effect.runPromise(
+			runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, { namespace: "pr-1" }, { dryRun: false }),
+		)
+		const strategy = (table: string) => report.tables.find((t) => t.table === table)?.strategy
+		expect(strategy("trace_list_mv")).toBe("filter")
+		expect(strategy("error_events")).toBe("rebuild")
+		expect(strategy("service_map_edges_hourly")).toBe("rebuild")
+
+		expect(count("SELECT toString(count()) AS n FROM traces WHERE ResourceAttributes['service.namespace'] = 'pr-1'")).toBe(0)
+		expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'api'")).toBe(1)
+		expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE ServiceNamespace = 'pr-1'")).toBe(0)
+		for (const [table, plan] of Object.entries(TABLE_DELETE_PLAN)) {
+			if (plan.strategy !== "scoped" || plan.service === null) continue
+			const where = plan.service.map((c) => `${c} = 'web'`).join(" OR ")
+			expect({ table, rows: count(`SELECT toString(count()) AS n FROM ${table} WHERE ${where}`) }).toEqual({
+				table,
+				rows: 0,
+			})
+		}
+		// Recomputed from pr-2's surviving api span, including the cascaded hourly rollup.
+		expect(count("SELECT toString(count()) AS n FROM error_events WHERE ServiceName = 'api'")).toBe(1)
+		expect(count("SELECT toString(sum(OccurrenceCount)) AS n FROM error_fingerprints_minutely")).toBe(1)
+		expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_minutely WHERE ServiceName = 'api'")).toBe(1)
+		expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_hourly WHERE ServiceName = 'api'")).toBe(1)
+		expect(count("SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'api'")).toBe(1)
+		expect(edge("web")).toBe(0)
+		expect(edge("caller")).toBe(1)
+		expect(existsSync(pendingDeletePath(dataDir))).toBe(false)
+	})
+})
diff --git a/docs/local-mode.md b/docs/local-mode.md
index 5bcfc26936..780fb00fc4 100644
--- a/docs/local-mode.md
+++ b/docs/local-mode.md
@@ -235,23 +235,27 @@ Checkpoints and `maple stop` wait for at most one hour's join. Sealed hours and
 retired UTC days are skipped. A span that arrives after its hour is sealed is
 not counted in the map's complete hours.
 
-### Deleting one service's data
+### Deleting a service's or namespace's data
 
-`maple delete` removes one service's telemetry from the live store without a
-full `maple reset`, for example when one shared local Maple serves many
-throwaway dev environments:
+`maple delete` removes one service's or one `service.namespace`'s telemetry from
+the live store without a full `maple reset`, for example when one shared local
+Maple serves many throwaway dev environments:
 
 ```bash
 maple delete --service checkout-pr-42                 # preview: per-table counts, exits 1
 maple delete --service checkout-pr-42 --yes           # delete everything for the service
+maple delete --namespace pr-42 --yes                  # every service in one service.namespace
+maple delete --service api --namespace pr-42 --yes    # one service within one namespace
 maple delete --service api --env preview-17 --yes     # one deployment.environment only
 maple delete --service api --before 7d --yes          # rows older than an age or UTC timestamp
 ```
 
-`--service` is required. `--env` matches `deployment.environment.name`, falling
-back to `deployment.environment`, the same expression the rollups store as
-`DeploymentEnv`. `--before` is floored to the UTC hour, because hourly rollups
-cannot split an hour; the report echoes the effective cutoff.
+Pass `--service`, `--namespace`, or both (a row must match all given). `--env`
+matches `deployment.environment.name`, falling back to `deployment.environment`,
+the same expression the rollups store as `DeploymentEnv`. `--before` is floored
+to the UTC hour, because hourly rollups cannot split an hour; the report echoes
+the effective cutoff. A namespace delete keeps a same-named service in another
+namespace: `api` in `pr-41` survives `--namespace pr-42`.
 
 The command is local-only: `--remote`, or a mode that resolves to remote, is
 refused before anything is sent. It needs a running `maple start`, reads the
@@ -262,32 +266,38 @@ inside the admission gate's exclusive section, so ingest, queries, the service
 map rollup and checkpoint backups wait for it.
 
 **Which tables.** Every table in the local schema is classified in
-`TABLE_DELETE_PLAN` (`apps/cli/src/server/scoped-delete.ts`), and
-`validateDeletePlan` checks the map against the bundled schema manifest. The
-server refuses to delete when they disagree, and
+`TABLE_DELETE_PLAN` (`apps/cli/src/server/scoped-delete.ts`). A table either is
+`excluded` or declares which of the three dimensions (service, namespace, env)
+it can filter exactly, plus how to recompute it when a request names one it
+lacks. `validateDeletePlan` checks the map against the bundled schema manifest
+for every request shape. The server refuses to delete when they disagree, and
 `apps/cli/test/scoped-delete.test.ts` fails when a table is added without a
-classification. There are three strategies:
-
-- **filter**: `ALTER TABLE ... DELETE` on the service, env and time columns. Used
-  for the six raw tables (env from `ResourceAttributes`) and the derived tables
-  that key on `ServiceName` and `DeploymentEnv`: `trace_list_mv`, `error_events*`,
-  `service_overview_*`, `service_operations_*`, `traces_aggregates_hourly`,
-  `trace_facets_hourly`, the service map tables, and so on. On a merging engine
-  (Aggregating, Summing, Replacing) every filter column must be in the sorting
-  key, so a row never mixes services. `service_map_edges_hourly` and
+classification. Per request, each table is handled one of three ways:
+
+- **filter**: the table declares every requested dimension, so
+  `ALTER TABLE ... DELETE` removes exactly the matching rows. The six raw tables
+  read namespace and env from `ResourceAttributes` and can always filter.
+  `trace_list_mv`, `service_overview_*`, `trace_facets_hourly`,
+  `logs_aggregates_hourly` and `trace_detail_spans` also carry the namespace;
+  most other rollups carry `ServiceName` and `DeploymentEnv` only. On a merging
+  engine (Aggregating, Summing, Replacing) every filter column must be in the
+  sorting key, so a row never mixes scopes. `service_map_edges_hourly` and
   `service_address_resolutions_hourly` match the service on either end of the
   edge, because an edge is derived from both services' spans.
-- **rebuild**: tables whose rows do not carry the service as a key.
-  `attribute_keys_hourly` and `attribute_values_hourly` have no service column,
-  and `error_fingerprints_minutely` stores `ServiceName` as an `anyLast`
-  aggregate. Tables that key on the service but have no environment
-  (`service_usage`, `metric_catalog`, `span_metrics_calls_hourly`,
-  `product_events`, `ai_crawler_requests`) use this path only under `--env`.
-  The hours the deleted source rows covered are cleared and recomputed by
-  re-running the table's own materialized-view bodies over the surviving source
-  rows, one UTC day per insert. An hour is recomputed only while every source
-  still holds all of it (bundled TTL plus a day of margin). Older hours are left
-  as they are and reported as skipped; their TTL removes them.
+- **rebuild**: the table lacks a requested dimension. The hours the deleted
+  source rows covered are cleared, limited to the requested dimensions the table
+  does key on, and recomputed from the surviving source rows. Most tables are
+  recomputed by re-running their own materialized-view bodies, one UTC day per
+  insert, in dependency order: a recomputed source fires its downstream views
+  (`service_operations_minutely` into `_hourly`, `error_events` into
+  `error_fingerprints_minutely`), and the downstream recompute that follows
+  wipes and redoes those hours. The two service map tables are recomputed by
+  re-running the hourly rollup from raw spans, only for hours it had already
+  sealed. `attribute_keys_hourly`, `attribute_values_hourly` and
+  `error_fingerprints_minutely` have no usable service key, so they are always
+  recomputed. An hour is recomputed only while every source still holds all of
+  it (bundled TTL plus a day of margin). Older hours are left as they are and
+  reported as skipped; their TTL removes them.
 - **excluded**: tables no local telemetry reaches. `session_events`,
   `session_replay_events`, `session_replays` and `identity_links` hold browser
   session data that local OTLP ingest never writes. `alert_checks` and
@@ -624,7 +634,7 @@ maple query 'SELECT ServiceName, count() FROM traces WHERE $__orgFilter AND $__t
   code, and no span id.
 - v2 lists cap at 100 rows per page; `maple error` caps sample traces at 20.
 - `maple delete` is local-only: it deletes from the local store (see
-  [Deleting one service's data](#deleting-one-services-data)) and refuses remote mode.
+  [Deleting a service's or namespace's data](#deleting-a-services-or-namespaces-data)) and refuses remote mode.
 
 ### Seeding data
 
diff --git a/docs/local-telemetry-archives.md b/docs/local-telemetry-archives.md
index 543f4e0c68..f444a7cec8 100644
--- a/docs/local-telemetry-archives.md
+++ b/docs/local-telemetry-archives.md
@@ -30,7 +30,7 @@ and the directory and manifest layouts.
 - Archive rehydration into the Maple UI. Historical data is queried in DuckDB,
   not reloaded into the dashboard.
 - Affected by `maple delete`. A scoped delete changes only the live store;
-  archived generations keep the deleted service's rows. Re-archiving an already
+  archived generations keep the deleted rows. Re-archiving an already
   archived day after a delete needs `--allow-shrink`.
 - A second always-running database. Archives are files; DuckDB opens them on
   demand.

From a23cf43e70837fe7ac148dba79e5ffd54459f4aa Mon Sep 17 00:00:00 2001
From: Makisuo <minetoaster12@gmail.com>
Date: Fri, 2 Oct 2026 01:05:14 +0200
Subject: [PATCH 3/4] fix(cli): send the maintenance token only to the store's
 own server

maple delete read the token beside whatever dataDir /local/status
claimed, so any server at the resolved URL could collect another store's
token. It now requires a loopback target and a live server that the
store's own discovery file names (same pid, dataDir and port).
---
 apps/cli/src/commands/delete.ts      | 47 +++++++++++++++++++++++++++-
 apps/cli/test/delete-command.test.ts | 36 ++++++++++++++++++++-
 2 files changed, 81 insertions(+), 2 deletions(-)

diff --git a/apps/cli/src/commands/delete.ts b/apps/cli/src/commands/delete.ts
index 569547d8a4..d8e9b64061 100644
--- a/apps/cli/src/commands/delete.ts
+++ b/apps/cli/src/commands/delete.ts
@@ -7,10 +7,12 @@ import { Mode } from "../core/mode"
 import { parseTimestampMs, sinceToMs } from "../core/time"
 import { CliUsageError } from "../lib/errors"
 import { bold, dim, green } from "../lib/style"
+import { isLoopbackHostname } from "../lib/local-address"
 import { maintenanceTokenPath } from "../server/archives/retention"
 import { hasSubject, ScopedDeleteReport, ScopedDeleteRequest } from "../server/scoped-delete"
 import { jsonFormatRequested, writeJson } from "./json-output"
-import { LocalStatus, prettyPath } from "./server-args"
+import { serverDiscoveryPath } from "../server/store-version"
+import { isProcessAlive, LocalStatus, prettyPath, ServerDiscovery } from "./server-args"
 
 const LOCAL_ONLY = "maple delete removes data from a local store only"
 
@@ -55,6 +57,48 @@ export const buildDeleteRequest = (flags: {
 const failure = (message: string, hint?: string) =>
 	hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint })
 
+const decodeDiscovery = Schema.decodeUnknownOption(Schema.fromJsonString(ServerDiscovery))
+
+/**
+ * The maintenance token goes only to the live loopback server its own store's
+ * discovery file names; any server could otherwise claim another store's dataDir.
+ */
+export const verifyStoreServer = (
+	baseUrl: string,
+	status: LocalStatus,
+	deps: {
+		readonly readDiscovery: (path: string) => string
+		readonly isAlive: (pid: number) => boolean
+	} = { readDiscovery: (path) => readFileSync(path, "utf8"), isAlive: isProcessAlive },
+): Effect.Effect<void, CliUsageError> =>
+	Effect.gen(function* () {
+		const target = URL.canParse(baseUrl) ? new URL(baseUrl) : undefined
+		if (target === undefined || !isLoopbackHostname(target.hostname.replace(/^\[(.*)\]$/, "$1")))
+			return yield* failure(
+				`maple delete only talks to a loopback server, not ${baseUrl}`,
+				"run it on the machine that runs `maple start`, against 127.0.0.1",
+			)
+		const path = serverDiscoveryPath(status.dataDir)
+		const discovery = Option.flatMap(
+			Option.liftThrowable(deps.readDiscovery)(path),
+			decodeDiscovery,
+		)
+		const port = (url: string) => (URL.canParse(url) ? new URL(url).port : undefined)
+		const matches = Option.exists(
+			discovery,
+			(found) =>
+				found.pid === status.pid &&
+				found.dataDir === status.dataDir &&
+				port(found.url) === target.port &&
+				deps.isAlive(found.pid),
+		)
+		if (!matches)
+			return yield* failure(
+				`the server at ${baseUrl} is not the live server ${prettyPath(path)} names`,
+				"point MAPLE_LOCAL_URL at the `maple start` that owns this store",
+			)
+	})
+
 const postDelete = (baseUrl: string, body: ScopedDeleteRequest & { readonly dryRun: boolean }) =>
 	Effect.gen(function* () {
 		const client = yield* HttpClient.HttpClient
@@ -63,6 +107,7 @@ const postDelete = (baseUrl: string, body: ScopedDeleteRequest & { readonly dryR
 			Effect.flatMap(HttpClientResponse.schemaBodyJson(LocalStatus)),
 			Effect.mapError(() => failure(`no Maple server answered at ${baseUrl}`, "start one with `maple start`")),
 		)
+		yield* verifyStoreServer(baseUrl, status)
 		// The token sits beside the data dir, so only the machine that owns the store can delete.
 		const tokenPath = maintenanceTokenPath(status.dataDir)
 		const token = yield* Effect.try({
diff --git a/apps/cli/test/delete-command.test.ts b/apps/cli/test/delete-command.test.ts
index 2777116b89..d35ba81633 100644
--- a/apps/cli/test/delete-command.test.ts
+++ b/apps/cli/test/delete-command.test.ts
@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, test } from "bun:test"
 import { Effect, Layer, Option } from "effect"
-import { buildDeleteRequest, parseBefore, resolveDeleteBaseUrl } from "../src/commands/delete"
+import { buildDeleteRequest, parseBefore, resolveDeleteBaseUrl, verifyStoreServer } from "../src/commands/delete"
 import { Mode } from "../src/core/mode"
 import { CliUsageError } from "../src/lib/errors"
 
@@ -67,3 +67,37 @@ describe("delete request", () => {
 		).toEqual(Option.some({ service: "api", namespace: "pr-42", env: "", beforeMs: 1_000 }))
 	})
 })
+
+describe("token goes only to the store's own server", () => {
+	const status = { service: "maple-local", pid: 4242, version: "x", url: "", dataDir: "/s/data", lastIngestAtMs: null } as const
+	const discovery = JSON.stringify({ pid: 4242, url: "http://127.0.0.1:4318", dataDir: "/s/data", startedAt: "t" })
+	const verify = (baseUrl: string, text: string | undefined, alive = true) =>
+		Effect.runPromise(
+			Effect.match(
+				verifyStoreServer(baseUrl, status, {
+					readDiscovery: () => {
+						if (text === undefined) throw new Error("ENOENT")
+						return text
+					},
+					isAlive: () => alive,
+				}),
+				{ onFailure: (error) => error.message, onSuccess: () => "ok" },
+			),
+		)
+
+	test("accepts the live loopback server the discovery file names", async () => {
+		expect(await verify("http://127.0.0.1:4318", discovery)).toBe("ok")
+		expect(await verify("http://localhost:4318", discovery)).toBe("ok")
+	})
+
+	test("refuses a non-loopback target", async () => {
+		expect(await verify("http://maple.home.arpa:4318", discovery)).toContain("only talks to a loopback server")
+	})
+
+	test("refuses a server that merely claims the store", async () => {
+		expect(await verify("http://127.0.0.1:9999", discovery)).toContain("is not the live server")
+		expect(await verify("http://127.0.0.1:4318", discovery.replace("4242", "1"))).toContain("is not the live server")
+		expect(await verify("http://127.0.0.1:4318", undefined)).toContain("is not the live server")
+		expect(await verify("http://127.0.0.1:4318", discovery, false)).toContain("is not the live server")
+	})
+})

From 3b2d58612ae03b61d0278f812d958f6234a5600b Mon Sep 17 00:00:00 2001
From: Makisuo <minetoaster12@gmail.com>
Date: Fri, 2 Oct 2026 01:13:42 +0200
Subject: [PATCH 4/4] fix(cli): report an out-of-range delete flag as invalid,
 not as a missing selector

---
 apps/cli/src/commands/delete.ts      | 40 +++++++++++++++++-----------
 apps/cli/test/delete-command.test.ts | 32 +++++++++++++++-------
 2 files changed, 47 insertions(+), 25 deletions(-)

diff --git a/apps/cli/src/commands/delete.ts b/apps/cli/src/commands/delete.ts
index d8e9b64061..7840febe39 100644
--- a/apps/cli/src/commands/delete.ts
+++ b/apps/cli/src/commands/delete.ts
@@ -38,24 +38,35 @@ export const parseBefore = (raw: string, nowMs: number): number | null => {
 	return age === null ? parseTimestampMs(raw) : nowMs - age
 }
 
+const failure = (message: string, hint?: string) =>
+	hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint })
+
 /** Present flags as a request; a blank `--service` or `--namespace` counts as absent. */
 export const buildDeleteRequest = (flags: {
 	readonly service: string | undefined
 	readonly namespace: string | undefined
 	readonly env: string | undefined
 	readonly beforeMs: number | undefined
-}): Option.Option<ScopedDeleteRequest> => {
-	const present = Object.entries({
-		service: flags.service?.trim() || undefined,
-		namespace: flags.namespace?.trim() || undefined,
-		env: flags.env,
-		beforeMs: flags.beforeMs === undefined ? undefined : Math.floor(flags.beforeMs),
-	}).filter(([, value]) => value !== undefined)
-	return Option.filter(Schema.decodeUnknownOption(ScopedDeleteRequest)(Object.fromEntries(present)), hasSubject)
-}
-
-const failure = (message: string, hint?: string) =>
-	hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint })
+}): Effect.Effect<ScopedDeleteRequest, CliUsageError> =>
+	Effect.gen(function* () {
+		const present = Object.entries({
+			service: flags.service?.trim() || undefined,
+			namespace: flags.namespace?.trim() || undefined,
+			env: flags.env,
+			beforeMs: flags.beforeMs === undefined ? undefined : Math.floor(flags.beforeMs),
+		}).filter(([, value]) => value !== undefined)
+		const request = yield* Schema.decodeUnknownEffect(ScopedDeleteRequest)(Object.fromEntries(present)).pipe(
+			Effect.mapError(() =>
+				failure(
+					"invalid delete flags",
+					"--service, --namespace and --env take at most 512 characters, and --before must be after 1970",
+				),
+			),
+		)
+		if (!hasSubject(request))
+			return yield* failure("pass --service, --namespace, or both", "e.g. maple delete --namespace pr-42")
+		return request
+	})
 
 const decodeDiscovery = Schema.decodeUnknownOption(Schema.fromJsonString(ServerDiscovery))
 
@@ -198,15 +209,12 @@ export const deleteCommand = Command.make("delete", {
 			const beforeMs = rawBefore === undefined ? undefined : parseBefore(rawBefore, Date.now())
 			if (beforeMs === null)
 				return yield* failure(`invalid --before: ${rawBefore}`, "use an age like 7d or a timestamp like 2026-10-01 12:00")
-			const built = buildDeleteRequest({
+			const request = yield* buildDeleteRequest({
 				service: Option.getOrUndefined(a.service),
 				namespace: Option.getOrUndefined(a.namespace),
 				env: Option.getOrUndefined(a.env),
 				beforeMs,
 			})
-			if (Option.isNone(built))
-				return yield* failure("pass --service, --namespace, or both", "e.g. maple delete --namespace pr-42")
-			const request = built.value
 			const report = yield* postDelete(baseUrl, { ...request, dryRun: !a.yes })
 			if (jsonFormatRequested()) yield* writeJson(report)
 			else
diff --git a/apps/cli/test/delete-command.test.ts b/apps/cli/test/delete-command.test.ts
index d35ba81633..c17a789411 100644
--- a/apps/cli/test/delete-command.test.ts
+++ b/apps/cli/test/delete-command.test.ts
@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, test } from "bun:test"
-import { Effect, Layer, Option } from "effect"
+import { Effect, Layer } from "effect"
 import { buildDeleteRequest, parseBefore, resolveDeleteBaseUrl, verifyStoreServer } from "../src/commands/delete"
 import { Mode } from "../src/core/mode"
 import { CliUsageError } from "../src/lib/errors"
@@ -54,17 +54,31 @@ describe("--before", () => {
 
 describe("delete request", () => {
 	const flags = { service: undefined, namespace: undefined, env: undefined, beforeMs: undefined }
+	const build = (input: Parameters<typeof buildDeleteRequest>[0]) =>
+		Effect.runPromise(
+			Effect.match(buildDeleteRequest(input), {
+				onFailure: (error) => ({ error: error.message }),
+				onSuccess: (request) => ({ request }),
+			}),
+		)
+
+	test("needs a service or a namespace", async () => {
+		expect(await build(flags)).toEqual({ error: "pass --service, --namespace, or both" })
+		expect(await build({ ...flags, service: "  ", env: "prod" })).toEqual({
+			error: "pass --service, --namespace, or both",
+		})
+	})
 
-	test("needs a service or a namespace", () => {
-		expect(Option.isNone(buildDeleteRequest(flags))).toBe(true)
-		expect(Option.isNone(buildDeleteRequest({ ...flags, service: "  ", env: "prod" }))).toBe(true)
+	test("reports an out-of-range value as invalid, not as a missing selector", async () => {
+		expect(await build({ ...flags, service: "a".repeat(513) })).toEqual({ error: "invalid delete flags" })
+		expect(await build({ ...flags, namespace: "pr-42", beforeMs: -1 })).toEqual({ error: "invalid delete flags" })
 	})
 
-	test("accepts a namespace alone or with a service, env and cutoff", () => {
-		expect(buildDeleteRequest({ ...flags, namespace: "pr-42" })).toEqual(Option.some({ namespace: "pr-42" }))
-		expect(
-			buildDeleteRequest({ service: " api ", namespace: "pr-42", env: "", beforeMs: 1_000.7 }),
-		).toEqual(Option.some({ service: "api", namespace: "pr-42", env: "", beforeMs: 1_000 }))
+	test("accepts a namespace alone or with a service, env and cutoff", async () => {
+		expect(await build({ ...flags, namespace: "pr-42" })).toEqual({ request: { namespace: "pr-42" } })
+		expect(await build({ service: " api ", namespace: "pr-42", env: "", beforeMs: 1_000.7 })).toEqual({
+			request: { service: "api", namespace: "pr-42", env: "", beforeMs: 1_000 },
+		})
 	})
 })