Track raw paths for ingest actions

packages/context/prompts/memory_agent_bundle_ingest_reconcile.md

@@ -11,10 +11,11 @@ Parsimonious. Stage 3 WUs already loaded `ingest_triage` and handled conflicts t
 2. Call `stage_list()` for the full index of this job's writes. If it is empty AND you have no evictions, exit — the runner short-circuits this case but the skill still teaches you to bail fast.
 3. If the system prompt includes `<canonical_pins>`, apply those pins before flagging a same-name or near-duplicate conflict. A pinned `canonicalArtifactKey` keeps the contested name when it is present in the Stage Index; competing variants keep or receive disambiguated names.
 4. Sweep both exact-key conflicts and near-duplicate writes. Compare WUs that wrote overlapping SL source names, overlapping wiki keys, the same `tables:` or `sl_refs:` action details, or obviously equivalent topic titles under different wiki keys. Call `stage_diff` to see the actual difference, and use `wiki_read`/`sl_read_source` when two different keys appear to describe the same table, metric, or source-of-truth mapping. If they're the same content, leave one canonical artifact and record the duplicate as subsumed. If they differ per `ingest_triage` rules, apply the correct resolution (rename + capture; election of canonical; silent replace for expression-only re-ingest change; or pinned canonical), then call `emit_conflict_resolution` with the artifact key and decision.
-5. Call `eviction_list()` for deleted raw paths. For each eviction: if inbound refs are empty, remove the artifact (`sl_delete`, `wiki_remove`); if inbound refs exist, retain with a deprecation marker. Then call `emit_eviction_decision` for every removed or retained artifact.
-6. If the Stage 4 sweep discovers a raw file whose only honest outcome is standalone SQL, wiki-only capture, or a human flag, call `emit_unmapped_fallback` with the raw path, reason, and fallback kind.
-7. Use `read_raw_span` to zoom into specific raw files when you need to resolve what two contested measures or wiki pages actually describe.
-8. Exit when you've processed every item.
+5. For any `wiki_write`, `wiki_remove`, `sl_write_source`, or `sl_edit_source` call you make during reconciliation, include `rawPaths` with only the raw paths that directly caused that reconciliation action.
+6. Call `eviction_list()` for deleted raw paths. For each eviction: if inbound refs are empty, remove the artifact (`sl_delete`, `wiki_remove`) and include that evicted raw path in `rawPaths`; if inbound refs exist, retain with a deprecation marker and include that evicted raw path in `rawPaths`. Then call `emit_eviction_decision` for every removed or retained artifact.
+7. If the Stage 4 sweep discovers a raw file whose only honest outcome is standalone SQL, wiki-only capture, or a human flag, call `emit_unmapped_fallback` with the raw path, reason, and fallback kind.
+8. Use `read_raw_span` to zoom into specific raw files when you need to resolve what two contested measures or wiki pages actually describe.
+9. Exit when you've processed every item.
 </workflow>
 
 <scope>

packages/context/prompts/memory_agent_bundle_ingest_work_unit.md

@@ -11,9 +11,10 @@ Assertive. The bundle was explicitly submitted for ingest. Default to capturing
 2. Load the per-source review skill first (e.g. `lookml_ingest`, `metricflow_ingest`, `dbt_ingest`), then `sl_capture` and `knowledge_capture`, and `ingest_triage` last. The triage skill tells you how to react when `wiki_sl_search` reveals that a prior WU already wrote something overlapping.
 3. If the system prompt includes `<canonical_pins>`, read those pins before choosing artifact keys. A pin's `canonicalArtifactKey` is the preferred artifact for its `contestedKey`: prefer editing the pinned canonical artifact when it already exists or when this raw file clearly updates it. Do not create a duplicate contested artifact when a pin says another artifact is canonical; use a specific disambiguated key only when the raw file describes a genuinely different domain.
 4. For each raw file: call `read_raw_file` (or `read_raw_span` for slicing large files) to load content. Before writing a new SL source or wiki page, call `wiki_sl_search` for each candidate name to find prior-WU writes; apply `ingest_triage` when you hit one, and apply any matching canonical pin before deciding whether to edit, rename, or skip.
-5. When `priorProvenance` names an existing artifact for one of your raw files, prefer `sl_edit` over `sl_write` for that artifact: the re-ingest change rule says expression-only changes replace silently, grain/column/filter changes replace and flag.
-6. When a raw file cannot map to normal SL and you use a fallback path, call `emit_unmapped_fallback` exactly once for that raw file and reason. Use `fallback: "sql_standalone"` for a standalone SQL source, `fallback: "wiki_only"` for documentation-only capture, and `fallback: "flagged"` when no reliable artifact can be written.
-7. When you're done, exit the loop without further tool calls.
+5. For every `wiki_write`, `wiki_remove`, `sl_write_source`, or `sl_edit_source` call, include `rawPaths` with only the raw file paths that directly support that action. If one artifact synthesizes several files, list each contributing raw file. Do not include unrelated files from the same WorkUnit.
+6. When `priorProvenance` names an existing artifact for one of your raw files, prefer `sl_edit` over `sl_write` for that artifact: the re-ingest change rule says expression-only changes replace silently, grain/column/filter changes replace and flag.
+7. When a raw file cannot map to normal SL and you use a fallback path, call `emit_unmapped_fallback` exactly once for that raw file and reason. Use `fallback: "sql_standalone"` for a standalone SQL source, `fallback: "wiki_only"` for documentation-only capture, and `fallback: "flagged"` when no reliable artifact can be written.
+8. When you're done, exit the loop without further tool calls.
 </workflow>
 
 <scope>

packages/context/skills/dbt_ingest/SKILL.md

@@ -27,6 +27,8 @@ For dbt context-source ingest, the dbt connection is usually not the warehouse c
 
 If a `models:` entry has no `columns:` block, or the available raw files do not confirm the physical column names, do **not** synthesize a full standalone source. Write a wiki note or a description-only overlay for the resolved manifest table instead. If a business metric is described but its referenced column is not confirmed in the warehouse schema, omit the measure and capture the unresolved intent in the wiki.
 
+Include `rawPaths` on every `wiki_write`, `sl_write_source`, and `sl_edit_source` call with only the dbt YAML files that directly support the action.
+
 After every `sl_write_source`, call `sl_validate`. A validation error saying a declared column or measure reference is absent from the physical table is a hard stop: re-read the warehouse-backed source and rewrite with confirmed names, or remove the invalid SL fields.
 
 ## 1.1 test hints (descriptions / meta)

packages/context/skills/knowledge_capture/SKILL.md

@@ -46,6 +46,8 @@ If nothing is worth capturing, respond without calling any tool.
 4. `wiki_write` to create or update. Prefer merging into an existing page over creating a new one.
 5. `wiki_remove` only when a page is truly obsolete — not to replace stale content (update it instead).
 
+For bundle/external ingest, include `rawPaths` on every `wiki_write`/`wiki_remove` call with only the raw files that directly support that wiki action. This keeps ingest provenance tied to the actual source file, not every file in the WorkUnit.
+
 ## Keys, summaries, and content
 
 - **Keys** are short kebab-case topic identifiers: `leads-source-filter`, `revenue-definition`, `churn-calculation`. No namespacing, no prefixes.

packages/context/skills/looker_ingest/SKILL.md

@@ -76,7 +76,7 @@ When `targetTable.ok === true`, the explore has a complete KTX backing target. B
 
 1. Use `targetTable.catalog`, `targetTable.schema`, and `targetTable.name` for `source_tables` preflight matching through `sl_discover` or `sl_read_source`.
 2. Use Looker field `sql`, labels, descriptions, and type metadata to derive source columns, measures, segments, joins, and grain.
-3. Call `sl_write_source` or `sl_edit_source` with `connectionId: targetWarehouseConnectionId`.
+3. Call `sl_write_source` or `sl_edit_source` with `connectionId: targetWarehouseConnectionId` and `rawPaths` set to the staged explore path.
 4. Set `source.name` to the deterministic API-derived source key, for example `looker__b2b__sales_pipeline`.
 5. Set `source.table` to `targetTable.canonicalTable`.
 6. Run `sl_validate` after every SL write.
@@ -85,13 +85,13 @@ The `table` field is `targetTable.canonicalTable`, not `rawSqlTableName`. Raw Lo
 
 Use `targetTable.{catalog,schema,name}` only for source_tables preflight. Do not put those tuple fields separately into the SL source unless the SL schema already asks for them.
 
-When `targetTable.ok === false`, keep the WU wiki-only for SL purposes. Capture durable domain semantics with `context_candidate_write`, then emit a fallback with the EXACT structured `reason` code from `targetTable.reason`. Put any human-readable context in `detail`, NOT in `reason`:
+When `targetTable.ok === false`, keep the WU wiki-only for SL purposes. Capture durable domain semantics with `context_candidate_write`, then emit a fallback with the EXACT structured `reason` code from `targetTable.reason`. Put any human-readable context in `clarification`, NOT in `reason`:
 
 ```json
 {
   "rawPath": "explores/b2b/sales_pipeline.json",
   "reason": "no_connection_mapping",
-  "detail": "Looker connection b2b_sandbox_bq is not mapped to a warehouse connection",
+  "clarification": "Looker connection b2b_sandbox_bq is not mapped to a warehouse connection",
   "fallback": "wiki_only"
 }
 ```

packages/context/skills/lookml_ingest/SKILL.md

@@ -43,6 +43,7 @@ When SL is allowed:
 - **Overlay** when the view is a thin wrapper over a manifest table (`sql_table_name:` matches a manifest entry). Do not repeat base columns or grain.
 - **Standalone** when the view uses `derived_table:` or `sql_always_where:`. `sl_write_source` rejects overlays whose name has no manifest entry; that error points here.
 - **Skip** a view with only `view:`, `sql_table_name:`, and bare `dimension:` entries (no `measure:`, `description:`, `derived_table:`, `sql_always_where:`, `join:`). The pre-filter already short-circuits those.
+- Include `rawPaths` on every `sl_write_source`/`sl_edit_source` call with the exact LookML raw file(s) that support the action.
 
 ## Preflight: never guess column names
 

packages/context/skills/metabase_ingest/SKILL.md

@@ -49,7 +49,8 @@ Use `resultMetadata` to:
 For each card:
 1. Analyze `resolvedSql` + `resultMetadata`: identify base tables, aggregations, joins, filters, column types.
 2. **REQUIRED before any write**: call `sl_discover` for every candidate target source name. The response tells you whether the name is manifest-backed (`Type: table` or `Type: sql`). For manifest-backed names you MUST use the overlay shape (`name:` + `measures:`/`segments:`/`description:` only — no `sql:`, `table:`, `grain:`, or `columns:`); the tool will reject a standalone write and you'll have wasted the call. If `sl_discover` returns nothing for the name, you can write a standalone source. Also call `sl_read_source` on existing sources you intend to extend so you don't duplicate measures.
-3. Decide:
+3. Include `rawPaths: ["cards/<id>.json"]` on every `sl_write_source`, `sl_edit_source`, and `wiki_write` call. If one artifact generalizes multiple near-duplicate cards, include each contributing card path and no unrelated cards.
+4. Decide:
    - Simple aggregation on a table that already has a source → `sl_edit_source` to add a measure.
    - Join between tables that should be linked in the SL graph → `sl_edit_source` to add a join.
    - Complex derived SQL (CTEs, multi-layer aggregation, scoring models) → `sl_write_source` with `source_type: sql`. When the SQL projects/filters from a single manifest-backed base table, set `inherits_columns_from: <manifest_key>` so columns inherit type and description from the manifest — see `sl_capture` skill for the slim form. Use `sl_discover` to discover the manifest key from the table reference in the SQL (it accepts `MARTS.CONSIGNMENTS`, `ANALYTICS.MARTS.CONSIGNMENTS`, or `CONSIGNMENTS`).
@@ -164,7 +165,7 @@ After Steps A and B, your SQL must:
 - Reference no aliases that aren't defined inside the SQL itself.
 - Be valid as a standalone subquery (the validator runs `SELECT * FROM (your_sql) LIMIT 1`).
 
-If `resolutionStatus: "fallback"` and the SQL is still complex enough that you can't confidently translate it, **skip the card** rather than writing broken SQL. Call `emit_unmapped_fallback` with the staged card path as `rawPath`, `reason: "parse_error"`, `detail: "metabase_sql_untranslated"`, and `fallback: "flagged"`.
+If `resolutionStatus: "fallback"` and the SQL is still complex enough that you can't confidently translate it, **skip the card** rather than writing broken SQL. Call `emit_unmapped_fallback` with the staged card path as `rawPath`, `reason: "parse_error"`, `clarification: "metabase_sql_untranslated"`, and `fallback: "flagged"`.
 
 ## Join-graph connectivity
 
@@ -175,7 +176,7 @@ For `source_type: table`:
 For `source_type: sql`:
 - The validator parses your SQL and rejects the write when a referenced manifest table has a viable projected local key but no declared `joins:` entry. Add the join only after confirming the output key and target grain match.
 - If `sl_discover` resolves the table, it is not outside the manifest. Do not write an `unmapped-table-*` fallback for resolved `orbit_raw`, `mart`, or other manifest-backed sources just because they appear inside card SQL.
-- If `sl_discover` cannot resolve a referenced table at all, write a single-line `wiki_write` with key `unmapped-table-<table_name>` so the gap is documented, then call `emit_unmapped_fallback` with the staged card path as `rawPath`, `reason: "missing_target_table"`, and `fallback: "wiki_only"`.
+- If `sl_discover` cannot resolve a referenced table at all, write a single-line `wiki_write` with key `unmapped-table-<table_name>` and `rawPaths: ["cards/<id>.json"]` so the gap is documented, then call `emit_unmapped_fallback` with the staged card path as `rawPath`, `reason: "missing_target_table"`, `tableRef: "<table_name>"`, and `fallback: "wiki_only"`. Do not use this fallback if `sl_discover` resolved the table/source.
 
 Joins on manifest-backed names compose: the manifest's joins are inherited automatically, and any overlay `joins:` are merged on top (deduped by `to` + `on`). Use `disable_joins: ["<on-clause>"]` in the overlay to suppress a specific manifest join. If `sl_discover` shows a manifest-backed source with `Joins: 0` and the warehouse FK metadata is genuinely absent, declaring application-level joins via the overlay is fair game — bootstrap with `sl_write_source` (overlay shape above), then refine via `sl_edit_source`.
 

packages/context/skills/notion_synthesize/SKILL.md

@@ -18,8 +18,8 @@ Each WorkUnit is either a single Notion page/span or a topical cluster of relate
 2. For each assigned page, call `read_raw_file`, or `read_raw_span` for oversized pages when the notes specify a span.
 3. Search `wiki_search` for existing pages that overlap the WorkUnit topics. Prefer updating an existing page over creating a duplicate.
 4. Use `context_evidence_search`, `context_evidence_read`, and `context_evidence_neighbors` to pull supporting chunks when indexed evidence is relevant. Pass `chunkId` and `documentId` values verbatim as returned by the evidence tools.
-5. Write durable business knowledge with `wiki_write`. Aim for a small number of high-quality pages per WorkUnit or cluster.
-6. When the Notion content defines a reusable dataset, metric, segment, join rule, source-of-truth mapping, or table with explicit columns, load `sl_capture`, discover existing sources first with `sl_discover` or `sl_read_source`, then use `sl_write_source` or `sl_edit_source` only for a confirmed mapped non-Notion target source. If no mapped target exists, call `emit_unmapped_fallback` and keep the content wiki-only.
+5. Write durable business knowledge with `wiki_write`. Aim for a small number of high-quality pages per WorkUnit or cluster. Include `rawPaths` with the exact Notion raw files that support each page.
+6. When the Notion content defines a reusable dataset, metric, segment, join rule, source-of-truth mapping, or table with explicit columns, load `sl_capture`, discover existing sources first with `sl_discover` or `sl_read_source`, then use `sl_write_source` or `sl_edit_source` only for a confirmed mapped non-Notion target source. Include `rawPaths` with the exact Notion raw files that support the SL action. If no mapped target exists, call `emit_unmapped_fallback` and keep the content wiki-only.
 7. For every deleted raw path in the Eviction Set, call `eviction_list`, decide retention, then `context_eviction_decision_write`. Do this even when no wiki write is needed.
 
 ## What To Capture
@@ -66,6 +66,7 @@ Search existing wiki pages for the same `tables:` or `sl_refs:` frontmatter and
 - Notion `dataSourceCount` counts Notion databases/data sources only. It does not prove that a warehouse/dbt table has or lacks a mapped semantic-layer source.
 - Do not create SL sources under the Notion connection just because a page mentions a warehouse, dbt, Looker, or Metabase object. Use the mapped warehouse/source connection after discovery, or emit an unmapped fallback and write wiki-only.
 - Distinguish fallback reasons precisely: if a non-Notion warehouse/dbt connection exists but `sl_discover` cannot find the named table/source, use `no_physical_table`; reserve `no_connection_mapping` for cases where there is no plausible non-Notion target connection at all.
+- If `sl_discover` resolves the table/source, do not call `emit_unmapped_fallback` for that table. Use the resolved source for `sl_refs`, overlay edits, or wiki-only documentation.
 - When calling `emit_unmapped_fallback`, pass the table or source identifier as `tableRef` (e.g. `tableRef: "orbit_analytics.customer"`) — the tool generates the canonical detail string from the reason code and `tableRef`. Use the optional `clarification` field only to add context that does not contradict the reason. Do not restate the reason in `clarification`.
 
 ## Tools

packages/context/skills/sl_capture/SKILL.md

@@ -190,6 +190,7 @@ use it) without changing its SQL expression or filters.
 - **`sl_edit_source`** is the workhorse for additive changes: add a measure, add a join, tweak a description, replace a filter. Cheap, targeted, preserves the rest of the file.
 - **`sl_write_source`** is for brand-new sources or when the entire file needs restructuring. It overwrites the file completely.
 - Do NOT modify existing measures or their descriptions unless the current turn explicitly corrects them.
+- During bundle/external ingest, include `rawPaths` on every `sl_write_source`/`sl_edit_source` call with only the raw files that directly support the SL action.
 
 ## Worked example — additive overlay
 

packages/context/src/ingest/ingest-bundle.runner.ts

@@ -5,7 +5,7 @@ import pLimit from 'p-limit';
 import { z } from 'zod';
 import { type KtxLogger, noopLogger } from '../core/index.js';
 import type { CaptureSession, MemoryAction } from '../memory/index.js';
-import type { SlValidationDeps } from '../sl/index.js';
+import type { SemanticLayerService, SemanticLayerSource, SlValidationDeps } from '../sl/index.js';
 import { createTouchedSlSources, type ToolContext, type ToolSession } from '../tools/index.js';
 import { actionTargetConnectionId } from './action-identity.js';
 import { NOTION_DEFAULT_MAX_KNOWLEDGE_CREATES_PER_RUN } from './adapters/notion/types.js';
@@ -86,6 +86,47 @@ function reportIdFromCreateResult(result: unknown): string | undefined {
   return typeof id === 'string' && id.length > 0 ? id : undefined;
 }
 
+function normalizeTableReference(value: string): string {
+  return value
+    .trim()
+    .replace(/["`]/g, '')
+    .replace(/[\[\]]/g, '')
+    .toLowerCase();
+}
+
+function finalReferenceSegment(value: string): string {
+  const parts = value.split('.').filter((part) => part.length > 0);
+  return parts.at(-1) ?? value;
+}
+
+function semanticSourceMatchesTableRef(source: SemanticLayerSource, tableRef: string): boolean {
+  const normalizedRef = normalizeTableReference(tableRef);
+  if (!normalizedRef) {
+    return false;
+  }
+
+  const normalizedSourceName = normalizeTableReference(source.name);
+  if (normalizedSourceName === normalizedRef) {
+    return true;
+  }
+
+  const table = typeof source.table === 'string' ? normalizeTableReference(source.table) : '';
+  if (table && table === normalizedRef) {
+    return true;
+  }
+
+  const refIsQualified = normalizedRef.includes('.');
+  if (refIsQualified && normalizedSourceName === finalReferenceSegment(normalizedRef)) {
+    return true;
+  }
+
+  return false;
+}
+
+function rawPathsForAction(action: MemoryAction, fallbackRawPaths: string[]): string[] {
+  return action.rawPaths && action.rawPaths.length > 0 ? [...new Set(action.rawPaths)] : fallbackRawPaths;
+}
+
 export class IngestBundleRunner {
   private readonly logger: KtxLogger;
   private readonly chainByConnection = new Map<string, Promise<unknown>>();
@@ -281,6 +322,24 @@ export class IngestBundleRunner {
     return blocks.join('\n\n');
   }
 
+  private async tableRefExistsInSemanticLayer(
+    semanticLayerService: SemanticLayerService,
+    connectionIds: string[],
+    tableRef: string,
+  ): Promise<boolean> {
+    for (const connectionId of connectionIds) {
+      try {
+        const sources = await semanticLayerService.loadAllSources(connectionId);
+        if (sources.some((source) => semanticSourceMatchesTableRef(source, tableRef))) {
+          return true;
+        }
+      } catch {
+        // Fallback diagnostics should not fail an ingest stage if an index lookup is temporarily unavailable.
+      }
+    }
+    return false;
+  }
+
   private resolveContextCuratorBudget(
     bundleRef: IngestBundleJob['bundleRef'],
     stageIndex: StageIndex,
@@ -603,6 +662,7 @@ export class IngestBundleRunner {
             preHead: sessionWorktree.baseSha,
             touchedSlSources: session.touchedSlSources,
             actions: sessionActions,
+            allowedRawPaths: new Set(wu.rawFiles),
             semanticLayerService: scopedSemanticLayerService,
             wikiService: scopedWikiService,
             configService: sessionWorktree.config,
@@ -667,6 +727,8 @@ export class IngestBundleRunner {
             emit_unmapped_fallback: createEmitUnmappedFallbackTool({
               stageIndex,
               allowedPaths: new Set(wu.rawFiles),
+              tableRefExists: (tableRef) =>
+                this.tableRefExistsInSemanticLayer(scopedSemanticLayerService, slConnectionIds, tableRef),
             }),
           };
 
@@ -825,6 +887,10 @@ export class IngestBundleRunner {
       const reconcileActions: MemoryAction[] = [];
       const rcScopedWiki = this.deps.wikiService.forWorktree(sessionWorktree.workdir);
       const rcScopedSl = this.deps.semanticLayerService.forWorktree(sessionWorktree.workdir);
+      const reconciliationAllowedRawPaths = new Set<string>([
+        ...currentHashes.keys(),
+        ...(eviction?.deletedRawPaths ?? []),
+      ]);
 
       const rcToolSession: ToolSession = {
         connectionId: job.connectionId,
@@ -832,6 +898,7 @@ export class IngestBundleRunner {
         preHead: reconcileSession.preHead,
         touchedSlSources: reconcileSession.touchedSlSources,
         actions: reconcileActions,
+        allowedRawPaths: reconciliationAllowedRawPaths,
         semanticLayerService: rcScopedSl,
         wikiService: rcScopedWiki,
         configService: sessionWorktree.config,
@@ -896,6 +963,7 @@ export class IngestBundleRunner {
         emit_unmapped_fallback: createEmitUnmappedFallbackTool({
           stageIndex,
           allowedPaths: allStagedPaths,
+          tableRefExists: (tableRef) => this.tableRefExistsInSemanticLayer(rcScopedSl, slConnectionIds, tableRef),
         }),
       };
 
@@ -1153,26 +1221,34 @@ export class IngestBundleRunner {
         return a.type === 'created' ? 'source_created' : 'measure_added';
       };
       const producedPaths = new Set<string>();
+      const pushActionProvenance = (rawPath: string, action: MemoryAction): void => {
+        const hash = currentHashes.get(rawPath) ?? 'unknown';
+        provenanceRows.push({
+          connectionId: job.connectionId,
+          sourceKey: job.sourceKey,
+          syncId,
+          rawPath,
+          rawContentHash: hash,
+          artifactKind: action.target,
+          artifactKey: action.key,
+          targetConnectionId: action.target === 'sl' ? actionTargetConnectionId(action, job.connectionId) : null,
+          artifactContentHash: null,
+          actionType: actionToType(action),
+        });
+        producedPaths.add(rawPath);
+      };
       for (const wu of stageIndex.workUnits) {
-        for (const rawPath of wu.rawFiles) {
-          const hash = currentHashes.get(rawPath) ?? 'unknown';
-          for (const action of wu.actions) {
-            provenanceRows.push({
-              connectionId: job.connectionId,
-              sourceKey: job.sourceKey,
-              syncId,
-              rawPath,
-              rawContentHash: hash,
-              artifactKind: action.target,
-              artifactKey: action.key,
-              targetConnectionId: action.target === 'sl' ? (action.targetConnectionId ?? null) : null,
-              artifactContentHash: null,
-              actionType: actionToType(action),
-            });
-            producedPaths.add(rawPath);
+        for (const action of wu.actions) {
+          for (const rawPath of rawPathsForAction(action, wu.rawFiles)) {
+            pushActionProvenance(rawPath, action);
           }
         }
       }
+      for (const action of reconcileActions) {
+        for (const rawPath of action.rawPaths ?? []) {
+          pushActionProvenance(rawPath, action);
+        }
+      }
       for (const resolution of stageIndex.artifactResolutions ?? []) {
         const hash = currentHashes.get(resolution.rawPath) ?? 'unknown';
         provenanceRows.push({

packages/context/src/ingest/local-bundle-ingest.test.ts

@@ -88,6 +88,35 @@ class WikiWritingAgentRunner extends AgentRunnerService {
   }
 }
 
+class WikiWritingWithRawPathAgentRunner extends AgentRunnerService {
+  override runLoop = vi.fn(async (params: any) => {
+    if (params.telemetryTags?.operationName === 'ingest-bundle-wu') {
+      const wikiWrite = params.toolSet.wiki_write;
+      if (!wikiWrite?.execute) {
+        throw new Error('wiki_write tool was not available to the WorkUnit');
+      }
+      const result = await wikiWrite.execute(
+        {
+          key: 'orders_context',
+          summary: 'Orders source context',
+          content: 'Orders are purchase records used for revenue analysis.',
+          tags: ['orders'],
+          rawPaths: ['orders/orders.json'],
+        },
+        { toolCallId: 'wiki-write' },
+      );
+      if (!result.structured.success) {
+        throw new Error(result.markdown);
+      }
+    }
+    return { stopReason: 'natural' as const };
+  });
+
+  constructor() {
+    super({ llmProvider: { getModel: () => ({}) as never } as never });
+  }
+}
+
 class HistoricSqlEvidenceAgentRunner extends AgentRunnerService {
   override runLoop = vi.fn(async (params: any) => {
     if (
@@ -374,6 +403,42 @@ describe('canonical local ingest', () => {
     }
   });
 
+  it('uses explicit action raw paths to avoid over-attributing work-unit provenance', async () => {
+    const sourceDir = join(tempDir, 'source');
+    await mkdir(join(sourceDir, 'orders'), { recursive: true });
+    await writeFile(join(sourceDir, 'orders', 'orders.json'), '{"name":"orders"}\n', 'utf-8');
+    await writeFile(join(sourceDir, 'orders', 'unrelated.json'), '{"name":"unrelated"}\n', 'utf-8');
+    const agentRunner = new WikiWritingWithRawPathAgentRunner();
+
+    const result = await runLocalIngest({
+      project,
+      adapters: [new FakeSourceAdapter()],
+      adapter: 'fake',
+      connectionId: 'warehouse',
+      sourceDir,
+      jobId: 'wiki-raw-path-local-1',
+      agentRunner,
+    });
+
+    expect(result.result.failedWorkUnits).toEqual([]);
+    expect(result.report.body.provenanceRows).toEqual([
+      {
+        rawPath: 'orders/orders.json',
+        artifactKind: 'wiki',
+        artifactKey: 'orders_context',
+        targetConnectionId: null,
+        actionType: 'wiki_written',
+      },
+      {
+        rawPath: 'orders/unrelated.json',
+        artifactKind: null,
+        artifactKey: null,
+        targetConnectionId: null,
+        actionType: 'skipped',
+      },
+    ]);
+  });
+
   it('runs historic-SQL evidence projection through the local bundle post-processor', async () => {
     const projectDir = join(tempDir, 'historic-sql-project');
     await initKtxProject({ projectDir, projectName: 'warehouse' });

packages/context/src/ingest/report-snapshot.ts

@@ -16,6 +16,7 @@ const ingestActionSchema = z.object({
   key: z.string(),
   detail: z.string(),
   targetConnectionId: z.string().nullable().default(null),
+  rawPaths: z.array(z.string()).optional(),
 });
 
 const touchedSlSourceSchema = z.object({

packages/context/src/ingest/tools/emit-reconciliation-records.tool.test.ts

@@ -201,6 +201,27 @@ describe('reconciliation emit tools', () => {
     expect(stageIndex.unmappedFallbacks).toEqual([]);
   });
 
+  it('rejects missing-table fallback decisions when the table resolves to an existing semantic source', async () => {
+    const stageIndex = makeStageIndex();
+    const tool = createEmitUnmappedFallbackTool({
+      stageIndex,
+      allowedPaths: new Set(['cards/revenue.json']),
+      tableRefExists: async (tableRef) => tableRef === 'orbit_analytics.mart_revenue_daily',
+    });
+
+    const output = await executeTool(tool, {
+      rawPath: 'cards/revenue.json',
+      reason: 'no_physical_table',
+      tableRef: 'orbit_analytics.mart_revenue_daily',
+      fallback: 'wiki_only',
+    });
+
+    expect(output).toContain(
+      'Error: tableRef "orbit_analytics.mart_revenue_daily" already resolves to a semantic source',
+    );
+    expect(stageIndex.unmappedFallbacks).toEqual([]);
+  });
+
   it('records explicit artifact resolutions for provenance rows', async () => {
     const stageIndex = makeStageIndex();
     const tool = createEmitArtifactResolutionTool({

packages/context/src/ingest/tools/emit-unmapped-fallback.tool.ts

@@ -5,6 +5,7 @@ import type { StageIndex, UnmappedFallbackRecord, UnmappedFallbackReason } from
 interface EmitUnmappedFallbackDeps {
   stageIndex: StageIndex;
   allowedPaths: ReadonlySet<string>;
+  tableRefExists?: (tableRef: string) => Promise<boolean>;
 }
 
 const unmappedFallbackReasonSchema = z.enum([
@@ -49,6 +50,10 @@ function canonicalDetail(reason: UnmappedFallbackReason, tableRef: string | unde
   }
 }
 
+function requiresMissingTableValidation(reason: UnmappedFallbackReason): boolean {
+  return reason === 'no_physical_table' || reason === 'missing_target_table';
+}
+
 export function createEmitUnmappedFallbackTool(deps: EmitUnmappedFallbackDeps) {
   return tool({
     description:
@@ -70,6 +75,12 @@ export function createEmitUnmappedFallbackTool(deps: EmitUnmappedFallbackDeps) {
       if (!deps.allowedPaths.has(input.rawPath)) {
         return `Error: rawPath "${input.rawPath}" is not available to this ingest stage`;
       }
+      if (input.tableRef && requiresMissingTableValidation(input.reason) && deps.tableRefExists) {
+        const exists = await deps.tableRefExists(input.tableRef);
+        if (exists) {
+          return `Error: tableRef "${input.tableRef}" already resolves to a semantic source; do not record ${input.reason} for an existing table.`;
+        }
+      }
 
       const base = canonicalDetail(input.reason, input.tableRef);
       const detail = input.clarification ? `${base} ${input.clarification.trim()}`.trim() : base;

packages/context/src/memory/types.ts

@@ -34,6 +34,7 @@ export interface MemoryAction {
   key: string;
   detail: string;
   targetConnectionId?: string | null;
+  rawPaths?: string[];
 }
 
 export interface MemoryAgentResult {

packages/context/src/sl/tools/sl-edit-source.tool.ts

@@ -1,6 +1,6 @@
 import YAML from 'yaml';
 import { z } from 'zod';
-import { addTouchedSlSource, type ToolContext, type ToolOutput } from '../../tools/index.js';
+import { addTouchedSlSource, type ToolContext, type ToolOutput, validateActionRawPaths } from '../../tools/index.js';
 import { applySqlEdits } from '../../tools/sql-edit-replacer.js';
 import { normalizeSemanticLayerDescriptions } from '../description-normalization.js';
 import type { SemanticLayerSource } from '../types.js';
@@ -25,6 +25,10 @@ const slEditSourceInputSchema = z.object({
     .optional()
     .describe('Targeted exact-match search/replace edits on the raw YAML content.'),
   delete: z.boolean().optional().describe('Set to true to delete this source entirely'),
+  rawPaths: z
+    .array(z.string().min(1))
+    .optional()
+    .describe('In ingest sessions, raw source file paths that directly support this SL action.'),
 });
 
 type SlEditSourceInput = z.infer<typeof slEditSourceInputSchema>;
@@ -75,6 +79,10 @@ If no source exists yet, use sl_write_source instead — this tool will reject t
 
     const semanticLayerService = context.session?.semanticLayerService ?? this.semanticLayerService;
     const skipIndex = context.session?.isWorktreeScoped === true;
+    const rawPathValidation = validateActionRawPaths(context.session, input.rawPaths);
+    if (!rawPathValidation.ok) {
+      return this.buildOutput(false, [rawPathValidation.error], sourceName);
+    }
 
     // Handle delete
     if (input.delete) {
@@ -88,6 +96,7 @@ If no source exists yet, use sl_write_source instead — this tool will reject t
             key: sourceName,
             detail: 'Deleted source',
             targetConnectionId: actionTargetConnectionId(context.session.connectionId, connectionId),
+            ...(rawPathValidation.rawPaths ? { rawPaths: rawPathValidation.rawPaths } : {}),
           });
         }
         return this.buildOutput(true, [], sourceName, { yaml: undefined, commitHash: undefined });
@@ -184,6 +193,7 @@ If no source exists yet, use sl_write_source instead — this tool will reject t
           key: sourceName,
           detail: `Applied ${editCount} edit(s)`,
           targetConnectionId: actionTargetConnectionId(context.session.connectionId, connectionId),
+          ...(rawPathValidation.rawPaths ? { rawPaths: rawPathValidation.rawPaths } : {}),
         });
       }
 

packages/context/src/sl/tools/sl-write-source.tool.ts

@@ -1,6 +1,6 @@
 import YAML from 'yaml';
 import { z } from 'zod';
-import { addTouchedSlSource, type ToolContext, type ToolOutput } from '../../tools/index.js';
+import { addTouchedSlSource, type ToolContext, type ToolOutput, validateActionRawPaths } from '../../tools/index.js';
 import { sourceOverlaySchema } from '../schemas.js';
 import type { SemanticLayerService } from '../semantic-layer.service.js';
 import type { SemanticLayerSource } from '../types.js';
@@ -25,6 +25,10 @@ const slWriteSourceInputSchema = z.object({
     .optional()
     .describe('Source definition (standalone with table/sql) or overlay (measures, computed columns, etc.)'),
   delete: z.boolean().optional().describe('Set to true to delete this source entirely'),
+  rawPaths: z
+    .array(z.string().min(1))
+    .optional()
+    .describe('In ingest sessions, raw source file paths that directly support this SL action.'),
 });
 
 type SlWriteSourceInput = z.infer<typeof slWriteSourceInputSchema>;
@@ -99,6 +103,10 @@ Do NOT join back to a table that the SQL already aggregates from if the grain co
 
     const semanticLayerService = context.session?.semanticLayerService ?? this.semanticLayerService;
     const skipIndex = context.session?.isWorktreeScoped === true;
+    const rawPathValidation = validateActionRawPaths(context.session, input.rawPaths);
+    if (!rawPathValidation.ok) {
+      return this.buildOutput(false, [rawPathValidation.error], sourceName);
+    }
 
     // Handle delete
     if (input.delete) {
@@ -116,6 +124,7 @@ Do NOT join back to a table that the SQL already aggregates from if the grain co
             key: sourceName,
             detail: 'Deleted source',
             targetConnectionId: actionTargetConnectionId(context.session.connectionId, connectionId),
+            ...(rawPathValidation.rawPaths ? { rawPaths: rawPathValidation.rawPaths } : {}),
           });
         }
         return this.buildOutput(true, [], sourceName, { yaml: undefined, commitHash: undefined });
@@ -142,6 +151,7 @@ Do NOT join back to a table that the SQL already aggregates from if the grain co
       context,
       semanticLayerService,
       skipIndex,
+      rawPathValidation.rawPaths,
     );
   }
 
@@ -154,6 +164,7 @@ Do NOT join back to a table that the SQL already aggregates from if the grain co
     context: ToolContext,
     semanticLayerService: SemanticLayerService,
     skipIndex: boolean,
+    rawPaths: string[] | undefined,
   ): Promise<ToolOutput<SemanticLayerStructured>> {
     const normalizedSource = normalizeSemanticLayerDescriptions(source, { fillMissing: !!context.session?.ingest });
     const isOverlay =
@@ -211,6 +222,7 @@ Do NOT join back to a table that the SQL already aggregates from if the grain co
           key: sourceName,
           detail: existing ? `Rewrote source` : `Created source`,
           targetConnectionId: actionTargetConnectionId(context.session.connectionId, connectionId),
+          ...(rawPaths ? { rawPaths } : {}),
         });
       }
 

packages/context/src/tools/action-raw-paths.ts

@@ -0,0 +1,30 @@
+import type { ToolSession } from './tool-session.js';
+
+type ActionRawPathValidation =
+  | { ok: true; rawPaths?: string[] }
+  | { ok: false; error: string };
+
+export function validateActionRawPaths(
+  session: ToolSession | undefined,
+  rawPaths: readonly string[] | undefined,
+): ActionRawPathValidation {
+  if (!rawPaths || rawPaths.length === 0) {
+    return { ok: true };
+  }
+
+  const uniqueRawPaths = [...new Set(rawPaths)];
+  const allowedRawPaths = session?.allowedRawPaths;
+  if (!allowedRawPaths) {
+    return { ok: true, rawPaths: uniqueRawPaths };
+  }
+
+  const unavailable = uniqueRawPaths.filter((rawPath) => !allowedRawPaths.has(rawPath));
+  if (unavailable.length > 0) {
+    return {
+      ok: false,
+      error: `rawPaths include unavailable ingest file(s): ${unavailable.join(', ')}`,
+    };
+  }
+
+  return { ok: true, rawPaths: uniqueRawPaths };
+}

packages/context/src/tools/index.ts

@@ -31,6 +31,7 @@ export { ingestMetadataRequired, resolveIngestMetadata } from './context-ingest-
 export type { SqlEdit } from './sql-edit-replacer.js';
 export { applySqlEdits } from './sql-edit-replacer.js';
 export type { IngestToolMetadata, MemoryAction, ToolSession } from './tool-session.js';
+export { validateActionRawPaths } from './action-raw-paths.js';
 export type { TouchedSlSource, TouchedSlSourceSet } from './touched-sl-sources.js';
 export {
   addTouchedSlSource,

packages/context/src/tools/tool-session.ts

@@ -16,6 +16,7 @@ export interface MemoryAction {
   key: string;
   detail: string;
   targetConnectionId?: string | null;
+  rawPaths?: string[];
 }
 
 interface EvictionDecisionRecord {
@@ -45,6 +46,7 @@ export interface ToolSession {
   preHead: string | null;
   touchedSlSources: TouchedSlSourceSet;
   actions: MemoryAction[];
+  allowedRawPaths?: ReadonlySet<string>;
   semanticLayerService: SemanticLayerService;
   wikiService: KnowledgeWikiService;
   configService: KtxFileStorePort;

packages/context/src/wiki/tools/wiki-remove.tool.ts

@@ -3,13 +3,17 @@ import type { KnowledgeIndexPort } from '../ports.js';
 import type { KnowledgeEventPort } from '../ports.js';
 type BlockScope = 'GLOBAL' | 'USER';
 import { KnowledgeWikiService } from '../index.js';
-import { BaseTool, type ToolContext, type ToolOutput } from '../../tools/index.js';
+import { BaseTool, type ToolContext, type ToolOutput, validateActionRawPaths } from '../../tools/index.js';
 
 const SYSTEM_AUTHOR = 'System User';
 const SYSTEM_EMAIL = 'system@example.com';
 
 const wikiRemoveInputSchema = z.object({
   key: z.string().describe('The page key to remove'),
+  rawPaths: z
+    .array(z.string().min(1))
+    .optional()
+    .describe('In ingest sessions, raw source file paths that directly support this removal.'),
 });
 
 type WikiRemoveInput = z.infer<typeof wikiRemoveInputSchema>;
@@ -42,6 +46,13 @@ export class WikiRemoveTool extends BaseTool<typeof wikiRemoveInputSchema> {
     const wikiService = context.session?.wikiService ?? this.wikiService;
     const writesGlobal = !!context.session;
     const skipIndex = context.session?.isWorktreeScoped === true;
+    const rawPathValidation = validateActionRawPaths(context.session, input.rawPaths);
+    if (!rawPathValidation.ok) {
+      return {
+        markdown: `Error: ${rawPathValidation.error}`,
+        structured: { success: false, key: input.key },
+      };
+    }
 
     const scope: BlockScope = writesGlobal ? 'GLOBAL' : 'USER';
     const scopeId = scope === 'USER' ? context.userId : null;
@@ -76,6 +87,7 @@ export class WikiRemoveTool extends BaseTool<typeof wikiRemoveInputSchema> {
         type: 'removed',
         key: input.key,
         detail: `Removed page "${input.key}"`,
+        ...(rawPathValidation.rawPaths ? { rawPaths: rawPathValidation.rawPaths } : {}),
       });
     }
 

packages/context/src/wiki/tools/wiki-write.tool.ts

@@ -4,7 +4,7 @@ import type { KnowledgeEventPort } from '../ports.js';
 type BlockScope = 'GLOBAL' | 'USER';
 import { KnowledgeWikiService, type WikiFrontmatter } from '../index.js';
 import { applySqlEdits } from '../../tools/sql-edit-replacer.js';
-import { BaseTool, type ToolContext, type ToolOutput } from '../../tools/index.js';
+import { BaseTool, type ToolContext, type ToolOutput, validateActionRawPaths } from '../../tools/index.js';
 
 const MAX_USER_BLOCKS = 100;
 const SYSTEM_AUTHOR = 'System User';
@@ -37,6 +37,10 @@ const wikiWriteInputSchema = z.object({
   representative_sql: z.string().optional(),
   usage: historicSqlUsageFrontmatterSchema.optional(),
   fingerprints: z.array(z.string()).optional(),
+  rawPaths: z
+    .array(z.string().min(1))
+    .optional()
+    .describe('In ingest sessions, raw source file paths that directly support this wiki action.'),
 });
 
 type WikiWriteInput = z.infer<typeof wikiWriteInputSchema>;
@@ -156,6 +160,13 @@ tags/refs/sl_refs use REPLACE semantics: omit to keep existing on update, [] to
     const wikiService = context.session?.wikiService ?? this.wikiService;
     const writesGlobal = !!context.session;
     const skipIndex = context.session?.isWorktreeScoped === true;
+    const rawPathValidation = validateActionRawPaths(context.session, input.rawPaths);
+    if (!rawPathValidation.ok) {
+      return {
+        markdown: `Error: ${rawPathValidation.error}`,
+        structured: { success: false, key: input.key },
+      };
+    }
 
     const scope: BlockScope = writesGlobal ? 'GLOBAL' : 'USER';
     const scopeId = scope === 'USER' ? context.userId : null;
@@ -261,7 +272,13 @@ tags/refs/sl_refs use REPLACE semantics: omit to keep existing on update, [] to
 
     const action = existing ? 'updated' : 'created';
     if (context.session) {
-      context.session.actions.push({ target: 'wiki', type: action, key: input.key, detail: input.summary });
+      context.session.actions.push({
+        target: 'wiki',
+        type: action,
+        key: input.key,
+        detail: input.summary,
+        ...(rawPathValidation.rawPaths ? { rawPaths: rawPathValidation.rawPaths } : {}),
+      });
     }
 
     // When the LLM used `replacements` (edit mode), it doesn't have the