Skip to content

Verify contributed source facts before graph assembly #394

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Kashkovsky merged 1 commit into from
Sep 7, 2026
Merged

Verify contributed source facts before graph assembly #394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/code_graph/checkpoint/commands.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ export interface CodeGraphCheckpointArtifactOptions {

export interface CodeGraphCheckpointExportOptions {
readonly cwd?: string;
/** @internal Bind publication to the physical snapshot whose assembly was verified. */
readonly expectedSnapshotId?: string;
readonly json?: boolean;
readonly output: string;
readonly quiet?: boolean;
Expand Down Expand Up @@ -164,6 +166,7 @@ export const runCodeGraphCheckpointExport = Effect.fn('codeGraph.checkpoint.expo
const snapshot = yield* store.readySnapshot(layout.databasePath, identity.worktreeId);
if (
snapshot === undefined ||
(options.expectedSnapshotId !== undefined && snapshot.id !== options.expectedSnapshotId) ||
snapshot.state !== 'ready' ||
snapshot.dirty ||
snapshot.commit !== identity.headCommit ||
Expand Down
49 changes: 30 additions & 19 deletions src/code_graph/indexer_build.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ import {
} from './indexer_snapshot_reuse.js';
import type {
CodeGraphIndexOptions,
CodeGraphSourceVerification,
CommittedBaseResult,
DirectPersistentCapacityProtection,
IncrementalOverlayAssessment,
Expand Down Expand Up @@ -245,6 +246,7 @@ export const buildOwnedCleanSnapshot = Effect.fn('codeGraph.buildOwnedCleanSnaps
readonly existing: CodeGraphSnapshot | undefined;
readonly fallbackSnapshotId: string;
readonly force: boolean;
readonly sourceVerification?: CodeGraphSourceVerification;
readonly fs: FileSystem.FileSystem;
readonly identity: RepositoryIdentity;
readonly inventory: CodeGraphInventory;
Expand Down Expand Up @@ -334,9 +336,10 @@ export const buildOwnedCleanSnapshot = Effect.fn('codeGraph.buildOwnedCleanSnaps
cleanFallbackAssessment = reused.value;
}
}
const resumed = input.force
? yield* input.store.resumableForcedBuild(input.layout.databasePath, input.logicalSnapshotId)
: undefined;
const resumed =
input.force && input.sourceVerification === undefined
? yield* input.store.resumableForcedBuild(input.layout.databasePath, input.logicalSnapshotId)
: undefined;
const building: CodeGraphSnapshot = resumed ?? {
commit: input.identity.headCommit,
dirty: false,
Expand Down Expand Up @@ -365,6 +368,7 @@ export const buildOwnedCleanSnapshot = Effect.fn('codeGraph.buildOwnedCleanSnaps
ensureVectors: input.ensureVectors,
existing: input.existing,
force: input.force,
sourceVerification: input.sourceVerification,
fs: input.fs,
identity: input.identity,
incrementalAssessment: cleanFallbackAssessment,
Expand Down Expand Up @@ -1005,6 +1009,7 @@ export const buildAndActivate = Effect.fn('codeGraph.buildAndActivate')(function
readonly embedding: CodeGraphEmbeddingIndexShape;
readonly ensureVectors: boolean;
readonly force: boolean;
readonly sourceVerification?: CodeGraphSourceVerification;
readonly fs: FileSystem.FileSystem;
readonly identity: RepositoryIdentity;
readonly inventory: CodeGraphInventory;
Expand Down Expand Up @@ -1439,21 +1444,22 @@ export const buildAndActivate = Effect.fn('codeGraph.buildAndActivate')(function
const loadingStartedAt = yield* Clock.currentTimeMillis;
// Attribution may inspect peer facts in this deterministic source batch (for example, TypeScript barrels).
// Reuse only a complete batch so a hit/miss partition cannot become a persisted derivation input.
const materializedShards = directPersistentMaterialization
? yield* input.store.loadMaterializedFileShards(
input.layout.databasePath,
files,
input.building.extractorSet,
shardDerivationIdentity,
{currentGraphContentId, snapshotIds: donorSnapshotIds},
)
: {
bytes: 0,
bytesByPath: new Map<string, number>(),
exactGenerationFiles: 0,
facts: new Map(),
materializedShardIdsByPath: new Map<string, string>(),
};
const materializedShards =
directPersistentMaterialization && input.sourceVerification === undefined
? yield* input.store.loadMaterializedFileShards(
input.layout.databasePath,
files,
input.building.extractorSet,
shardDerivationIdentity,
{currentGraphContentId, snapshotIds: donorSnapshotIds},
)
: {
bytes: 0,
bytesByPath: new Map<string, number>(),
exactGenerationFiles: 0,
facts: new Map(),
materializedShardIdsByPath: new Map<string, string>(),
};
const exactGenerationShardFiles = materializedShards.exactGenerationFiles;
const materializedShardBatchComplete =
directPersistentMaterialization &&
Expand Down Expand Up @@ -1511,10 +1517,15 @@ export const buildAndActivate = Effect.fn('codeGraph.buildAndActivate')(function
unit: 'files',
}) ?? Effect.void;
const attributionStartedAt = yield* Clock.currentTimeMillis;
// Original worker objects flow directly into this assembly; they never become raw-cache donors.
const materializationFacts =
input.sourceVerification === undefined
? cached.facts
: yield* input.sourceVerification.materializeFacts({facts: cached.facts, files: fallbackFiles});
let flushShardCacheAfterAttribution = false;
const attributedFallbackFacts = materializationSubphases.measure('attributionCompute', () =>
attributeFacts(
fallbackFiles.map(file => input.languagePacks.postprocessFile(file, cached.facts.get(file.path)!)),
fallbackFiles.map(file => input.languagePacks.postprocessFile(file, materializationFacts.get(file.path)!)),
),
);
replayMetrics = addMaterializationReplayMetrics(replayMetrics, {
Expand Down
7 changes: 7 additions & 0 deletions src/code_graph/indexer_materialization.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,12 @@ function codeGraphFileProgressDimensions(
export function cacheContentBatch(options: {
readonly databasePath: string;
readonly languagePacks: CodeGraphLanguagePackRegistryShape;
/** Required verification runs before persistence; unlike contribution enqueue, failures propagate. */
readonly onSourceParserBatch?: (group: {
readonly cacheIdentity: string;
readonly facts: readonly BoundedCodeGraphFact[];
readonly files: readonly CodeGraphInventoryFile[];
}) => Effect.Effect<void, unknown>;
readonly onCachedParserBatch?: (group: {
readonly cacheIdentity: string;
readonly facts: readonly BoundedCodeGraphFact[];
Expand Down Expand Up @@ -326,6 +332,7 @@ export function cacheContentBatch(options: {
currentScanningMetrics(),
);
const startedAt = performance.now();
yield* options.onSourceParserBatch?.(group) ?? Effect.void;
yield* options.store.cacheFacts(
options.databasePath,
group.files,
Expand Down
120 changes: 75 additions & 45 deletions src/code_graph/indexer_service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,17 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
request.threadnoteHome,
).pipe(Effect.provideService(Crypto.Crypto, crypto));
const requestedOverlay = requestedBuildRequest.state;
if (
request.sourceVerification &&
(request.force !== true ||
request.sourceOnly !== true ||
request.includeOverlay !== false ||
requestedOverlay.dirty)
) {
return yield* CodeGraphIndexOperationError.make({
message: 'Source verification requires a clean, forced source-only build without overlays.',
});
}
yield* anonymousTelemetry.observeOverlay(requestedOverlay.dirty);
const ensureVectors = codeGraphIndexEnsuresVectors(request);
const requestKey = request.force
Expand Down Expand Up @@ -197,7 +208,7 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
Effect.andThen(request.onProgress?.(progress) ?? Effect.void),
),
};
if (yield* fs.exists(graphShareEnrollmentPath(path, initialIdentity.repoRoot))) {
if (!options.sourceOnly && (yield* fs.exists(graphShareEnrollmentPath(path, initialIdentity.repoRoot)))) {
yield* captureSharedGraphImportBase({
cwd: request.cwd,
identity: initialIdentity,
Expand All @@ -215,20 +226,21 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
temporaryDirectory: system.tempDirectory,
walAutoCheckpointPages: options.sqliteWriterTuning?.walAutoCheckpointPages ?? 1_000,
};
yield* hydrateSharedParseCache({
databasePath: layout.databasePath,
identity: initialIdentity,
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
capacityProtection,
fs,
if (!options.sourceOnly)
yield* hydrateSharedParseCache({
databasePath: layout.databasePath,
identity: initialIdentity,
layout,
onProgress: options.onProgress,
threadnoteHome: options.threadnoteHome,
}),
store,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
capacityProtection,
fs,
identity: initialIdentity,
layout,
onProgress: options.onProgress,
threadnoteHome: options.threadnoteHome,
}),
store,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
const repositoryBuild = withCodeGraphProcessLock(
fs,
layout.lockPath,
Expand Down Expand Up @@ -347,7 +359,10 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
const cacheCoalescer = cacheContentBatch({
databasePath: layout.databasePath,
languagePacks,
onCachedParserBatch: group => enqueueSharedParserBatch(identity, options.threadnoteHome, group),
onSourceParserBatch: options.sourceVerification?.observeParserBatch,
onCachedParserBatch: options.sourceOnly
? undefined
: group => enqueueSharedParserBatch(identity, options.threadnoteHome, group),
onProgress: options.onProgress,
parserPool,
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
Expand Down Expand Up @@ -492,9 +507,10 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
: undefined;
const forcedSnapshotId = forcedSnapshotIdentity(logicalSnapshotId, forceGeneration);
const directSnapshotId = directFullSnapshotIdentity(logicalSnapshotId);
const resumedForcedBuild = options.force
? yield* store.resumableForcedBuild(layout.databasePath, logicalSnapshotId)
: undefined;
const resumedForcedBuild =
options.force && options.sourceVerification === undefined
? yield* store.resumableForcedBuild(layout.databasePath, logicalSnapshotId)
: undefined;
const readyCandidateIds = inventory.dirty
? options.incrementalOverlay === false
? [directSnapshotId]
Expand Down Expand Up @@ -586,6 +602,7 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
existing,
fallbackSnapshotId: forcedSnapshotId,
force: options.force === true,
sourceVerification: options.sourceVerification,
fs,
identity,
inventory,
Expand Down Expand Up @@ -946,10 +963,11 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
}).pipe(Effect.ignore),
),
);
yield* drainQueuedGraphShareContributions({
identity: initialIdentity,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
if (!options.sourceOnly)
yield* drainQueuedGraphShareContributions({
identity: initialIdentity,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
return summary;
}),
).pipe(
Expand All @@ -963,11 +981,15 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
Effect.provideService(CodeGraphMaintenanceCoordinator, maintenance),
Effect.provideService(HttpClient.HttpClient, http),
Effect.catchIf(
cause => Schema.is(WorktreeChangedDuringIndex)(cause) && attempt === 0,
cause =>
request.sourceVerification === undefined && Schema.is(WorktreeChangedDuringIndex)(cause) && attempt === 0,
() => indexAttempt(request, anonymousTelemetry, attempt + 1, bypassCachedFacts),
),
Effect.catchIf(
cause => Schema.is(CachedCodeGraphFactUnavailableDuringIndex)(cause) && !bypassCachedFacts,
cause =>
request.sourceVerification === undefined &&
Schema.is(CachedCodeGraphFactUnavailableDuringIndex)(cause) &&
!bypassCachedFacts,
() => indexAttempt(request, anonymousTelemetry, attempt, true),
),
);
Expand All @@ -980,7 +1002,9 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
withCodeGraphBuildAnonymousTelemetry(anonymousTelemetry, indexAttempt(request, anonymousTelemetry)),
);
const ensureCommitWithSummary = (
request: Omit<CodeGraphIndexOptions, 'force' | 'includeOverlay'> & {readonly commit: string},
request: Omit<CodeGraphIndexOptions, 'force' | 'includeOverlay' | 'sourceVerification'> & {
readonly commit: string;
},
anonymousTelemetry: CodeGraphBuildAnonymousTelemetryReporter,
bypassCachedFacts = false,
): Effect.Effect<{readonly lease: CodeGraphCommitLease; readonly summary: CodeGraphIndexSummary}, unknown> =>
Expand Down Expand Up @@ -1035,28 +1059,29 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
temporaryDirectory: system.tempDirectory,
walAutoCheckpointPages: options.sqliteWriterTuning?.walAutoCheckpointPages ?? 1_000,
};
if (yield* fs.exists(graphShareEnrollmentPath(path, initialIdentity.repoRoot))) {
if (!options.sourceOnly && (yield* fs.exists(graphShareEnrollmentPath(path, initialIdentity.repoRoot)))) {
yield* captureSharedGraphImportBase({
cwd: request.cwd,
identity: commitIdentity,
onProgress: options.onProgress,
threadnoteHome: request.threadnoteHome,
});
}
yield* hydrateSharedParseCache({
databasePath: layout.databasePath,
identity: commitIdentity,
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
capacityProtection,
fs,
if (!options.sourceOnly)
yield* hydrateSharedParseCache({
databasePath: layout.databasePath,
identity: commitIdentity,
layout,
onProgress: options.onProgress,
threadnoteHome: options.threadnoteHome,
}),
store,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
capacityProtection,
fs,
identity: commitIdentity,
layout,
onProgress: options.onProgress,
threadnoteHome: options.threadnoteHome,
}),
store,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
const commitBuild = withCodeGraphProcessLock(
fs,
layout.lockPath,
Expand Down Expand Up @@ -1115,7 +1140,9 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
const cacheCoalescer = cacheContentBatch({
databasePath: layout.databasePath,
languagePacks,
onCachedParserBatch: group => enqueueSharedParserBatch(identity, options.threadnoteHome, group),
onCachedParserBatch: options.sourceOnly
? undefined
: group => enqueueSharedParserBatch(identity, options.threadnoteHome, group),
onProgress: options.onProgress,
parserPool,
persistentCapacityProtector: codeGraphDirectPersistentCapacityProtector({
Expand Down Expand Up @@ -1202,10 +1229,11 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
}).pipe(Effect.ignore),
),
);
yield* drainQueuedGraphShareContributions({
identity: initialIdentity,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
if (!options.sourceOnly)
yield* drainQueuedGraphShareContributions({
identity: initialIdentity,
threadnoteHome: request.threadnoteHome,
}).pipe(Effect.ignore);
return lease;
}),
).pipe(
Expand All @@ -1224,7 +1252,9 @@ export class CodeGraphIndexer extends Context.Service<CodeGraphIndexer, CodeGrap
),
);
const ensureCommit = (
request: Omit<CodeGraphIndexOptions, 'force' | 'includeOverlay'> & {readonly commit: string},
request: Omit<CodeGraphIndexOptions, 'force' | 'includeOverlay' | 'sourceVerification'> & {
readonly commit: string;
},
) =>
Effect.flatMap(
makeCodeGraphBuildAnonymousTelemetryReporter(
Expand Down
Loading