//! Delivery-layer orchestration shared by the `Code System Graph` CLI and MCP server.
pub mod http_server;
pub mod mcp;
mod sync;
mod work_state;
mod worker;
use std::collections::{BTreeMap, BTreeSet};
use std::ffi::OsStr;
use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use atomic_write_file::AtomicWriteFile;
use code_system_graph_core::{
AffectedTestsRequest, AnalyzerVersions, ArtifactKey, BatchAction, BatchPlanError, BitbucketProvider, ChangeAnalysisError, ChangeAnalysisOptions, ChangeError, ChangeImpactReport, ChangeProvider, ChangeRequest, ChangeScope, ChangeSet, CodeGraphConfig, CodeGraphProvider, CommunityError, ConfigDoctorInput, ConfigError, ConfigExtractionError, ContractReport, ContractRequest, CorroborationReport, DataDocument, DataExtractionError, DeclaredImplementation, DeclaredTestCase, DoctorReport, DoctorRequest, DocumentationDocument, DocumentationExtractionError, EXTRACTION_CONTRACT_VERSION, EffectiveRepositoryConfig, EventDocument, EventExtractionError, EventGraphFacts, ExecutionPolicy, ExitCode, ExportReport, ExportRequest, ExtractionBudgets, ExtractionGraphFacts, ExtractionLimitExceeded, ExtractionTracker, ExtractorBatch, ExtractorBatchPlan, FederatedGraph, FreshnessDoctorInput, GeneratedClientError, GeneratedClientMetadata, GitCliChangeProvider, GitHubProvider, GraphqlDocument, GraphqlExtractionError, GraphqlGraphFacts, HttpBoundary, HttpExtractionError, ImpactContext, ImpactError, ImpactReport, ImpactRequest, ImpactTarget, IncrementalPlan, InfrastructureDocument, InfrastructureExtractionError, IntegrityDoctorInput, InterfaceError, LinkError, LocalCodeIntelligenceProvider, LocalContextRequest, LocalContextResult, LocalEnrichmentInput, LocalEnrichmentStatus, LocalImpactItem, LocalImpactRequest, ManifestEdit, ManifestEditError, ManifestError, ManualLinkConfig, ManualLinkError, PackageGraphFacts, PackageManifest, PackageManifestError, PrAuthToken, ProtobufDocument, ProtobufExtractionError, ProtobufGraphFacts, ProviderBudget, ProviderCapability, ProviderDoctorInput, ProviderDoctorStatus, ProviderError, ProviderRequest, ProviderStatus, PullRequestCoordinates, PullRequestError, PullRequestInspectRequest, PullRequestInspection, PullRequestListPage, PullRequestListRequest, PullRequestListState, PullRequestProvider, PullRequestProviderConfig, PullRequestProviderKind, QueryError, RecommendedCommand, RegisteredWorkspace, RegistryError, ReqwestPrHttpTransport, SafeConfigDocument, SchemaDoctorInput, SearchFilters, SearchReport, SearchRequest, SourceEpistemicStatus, SourceGraphFacts, SourceLanguage, SourceObservation, SourceRole, SourceSyntaxError, SourceSyntaxLanguage, SourceWarning, SymbolAnchor, SymbolCorroboration, TraceError, TraversalReport, TraversalRequest, WorkspaceManifest, affected_link_keys, analyze_changes, analyze_communities_with_progress, analyze_impact, apply_openapi_override, classify_interface_error, commit_manifest_edit, compare_community_snapshots, corroborate_repository, declared_implementation, declared_test_case, doctor, documents_to_graph, encode_native_path, event_documents_to_graph, export_graph, extract_asyncapi, extract_codeowners, extract_data_artifact, extract_docker_compose, extract_generated_client_metadata, extract_graphql_document_with_tracker, extract_graphql_persisted_operations_with_tracker, extract_helm, extract_kubernetes, extract_markdown, extract_openapi_with_tracker, extract_package_manifest_with_tracker, extract_protobuf_with_tracker, extract_safe_config, extract_service_catalog, extract_terraform, graphql_documents_to_graph, inspect_contracts, inspect_source_syntax, link_declared_implementations_with_ambiguities, link_declared_tests_with_ambiguities, link_http_boundaries_with_ambiguities, link_registered_package_owners, load_extractor_batch_with_budgets, merge_affected_link_neighborhoods, package_manifest_to_graph, parse_event_source, parse_go_source_with_tracker, parse_graphql_source_with_tracker, parse_java_source_with_tracker, parse_javascript_source_at_path_with_tracker, parse_literal_sql_source_at_root, parse_manifest, parse_protobuf_generated_source, parse_python_source_with_tracker, parse_rust_source_with_tracker, parse_typescript_source_at_path_with_tracker, plan_extractor_batches, plan_incremental_scan, precheck_focused_source_values, preview_add_manual_link, preview_add_repository, preview_remove_repository, protobuf_documents_to_graph, register_workspace, resolve_manual_links, resolve_repository_config, search, source_observations_to_graph, store_extractor_batch, traverse
};
pub use code_system_graph_core::{
ConfigSource, DEFAULT_EXCLUDES, IgnorePolicy, PROTECTED_EXCLUDES
};
use code_system_graph_model::{
ArtifactFingerprint, CheckoutId, Community, CommunityAlgorithm, CommunityConfig, CommunityDelta, CommunityId, CommunityScope, Edge, EdgeId, EdgeKind, EpistemicStatus, Evidence, EvidenceId, ExtractorRun, ExtractorRunStatus, FreshnessSummary, LinkDecision, LinkStatus, Node, NodeId, NodeKind, OverallFreshness, Provenance, RepoFreshness, RepoFreshnessState, RepoId, RepositoryRecord, StoredExtractorBatch, ToolEnvelope, ToolStatus, TraceReport, WorkspaceRecord, stable_id, stable_id_bytes
};
use code_system_graph_store_sqlite::{
ManualLinkDisposition, ManualLinkRecord, ProviderCapabilityRecord, QueryCacheRecord, SnapshotBatch, SqliteStore, StoreError, StoreLock, latest_schema_version
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
pub use sync::{
CodeGraphRepositorySync, CodeGraphSyncState, CodeGraphSyncSummary, SyncSummary, SyncTarget, load_persisted_watch_targets, sync_workspace_with_overrides, sync_workspace_with_wall_time_cap, sync_workspace_with_worker_executable, workspace_sync_targets
};
use thiserror::Error;
use tokio_util::sync::CancellationToken;
#[doc(hidden)]
pub use worker::{
SupervisedProcessGroup, configure_supervised_process_group, run_worker_from_stdio, terminate_process_tree, terminate_supervised_process
};
const MAX_TRACE_DEPTH: usize = 32;
const MAX_SCAN_DEGRADATIONS: usize = 25;
const GENERATED_STATE_IGNORE_RULE: &[u8] = b".code-system-graph/";
pub(crate) const CODEGRAPH_DISABLED_CODE: &str = "codegraph_disabled";
pub(crate) const CODEGRAPH_DISABLED_MESSAGE: &str =
"CodeGraph is disabled by the trusted process policy.";
/// Error returned while executing delivery-layer application operations.
#[derive(Debug, Error)]
pub enum ApplicationError {
/// File input could not be read.
#[error("failed to read `{path}`: {source}")]
ReadFile {
/// Path that could not be read.
path: PathBuf,
/// Underlying operating-system error.
source: std::io::Error,
},
/// Explicit output could not be created without overwriting existing data.
#[error("failed to create `{path}`: {source}")]
WriteFile {
/// Path that could not be created.
path: PathBuf,
/// Underlying operating-system error.
source: std::io::Error,
},
/// Workspace manifest was invalid.
#[error(transparent)]
Manifest(#[from] ManifestError),
/// An analyzed repository attempted to control advanced global resource policy.
#[error(
"advanced global resource policy in `{config}` is inside analyzed repository `{repository}`; move the workspace manifest outside every analyzed checkout"
)]
UntrustedGlobalPolicySource {
/// Canonical workspace manifest path.
config: PathBuf,
/// Manifest alias whose checkout contains the policy source.
repository: String,
},
/// An extractor exhausted one configured per-invocation resource.
#[error(transparent)]
ExtractionLimit(#[from] ExtractionLimitExceeded),
/// A supervised pass exhausted one configured execution resource.
#[error(transparent)]
ExecutionLimit(#[from] code_system_graph_core::ExecutionLimitExceeded),
/// HTTP boundary artifact was invalid.
#[error(transparent)]
HttpExtraction(#[from] HttpExtractionError),
/// Focused package manifest was malformed.
#[error(transparent)]
PackageManifest(#[from] PackageManifestError),
/// Generated-client metadata was malformed or unsafe.
#[error(transparent)]
GeneratedClient(#[from] GeneratedClientError),
/// GraphQL contract or persisted-operation metadata was malformed.
#[error(transparent)]
Graphql(#[from] GraphqlExtractionError),
/// `AsyncAPI` event contract was malformed or unsupported.
#[error(transparent)]
Event(#[from] EventExtractionError),
/// Protobuf contract was malformed or internally inconsistent.
#[error(transparent)]
Protobuf(#[from] ProtobufExtractionError),
/// Database or SQL artifact was malformed or unsupported.
#[error(transparent)]
Data(#[from] DataExtractionError),
/// Infrastructure or deployment artifact was malformed or unsupported.
#[error(transparent)]
Infrastructure(#[from] InfrastructureExtractionError),
/// Documentation, catalog, or ownership artifact was malformed.
#[error(transparent)]
Documentation(#[from] DocumentationExtractionError),
/// Secret-safe configuration key extraction failed.
#[error(transparent)]
ConfigExtraction(#[from] ConfigExtractionError),
/// Mandatory source grammar inspection failed.
#[error(transparent)]
SourceSyntax(#[from] SourceSyntaxError),
/// A focused parser emitted a fact without structural syntax corroboration.
#[error("invalid focused source observation: {0}")]
InvalidSourceObservation(String),
/// Incremental extractor batch state was inconsistent.
#[error(transparent)]
BatchPlan(#[from] BatchPlanError),
/// Deterministic linking failed.
#[error(transparent)]
Link(#[from] LinkError),
/// Exact manual relationship resolution failed.
#[error(transparent)]
ManualLink(#[from] ManualLinkError),
/// Repository registry validation failed.
#[error(transparent)]
Registry(#[from] RegistryError),
/// Effective repository configuration could not be resolved.
#[error(transparent)]
Config(#[from] ConfigError),
/// Workspace manifest mutation failed.
#[error(transparent)]
ManifestEdit(#[from] ManifestEditError),
/// A validated registry unexpectedly omitted a manifest alias.
#[error("validated registry omitted repository alias `{0}`")]
RegistryAliasMissing(String),
/// CLI override references an unknown repository alias.
#[error("CLI override references unknown repository alias `{0}`")]
UnknownOverrideRepository(String),
/// Requested workspace name differs from the manifest.
#[error("workspace name `{requested}` does not match manifest name `{manifest}`")]
WorkspaceNameMismatch {
/// Name supplied by the caller.
requested: String,
/// Name declared by the manifest.
manifest: String,
},
/// Workspace already exists in the selected registry.
#[error("workspace `{0}` is already registered")]
WorkspaceAlreadyExists(String),
/// Another verified watcher owns the workspace lease.
#[error("watcher for workspace `{0}` is already active")]
WatcherAlreadyActive(String),
/// Workspace is absent from the selected registry.
#[error("workspace `{0}` is not registered")]
WorkspaceNotFound(String),
/// A targeted scan cannot mix batches produced under different global budgets.
#[error(
"extraction budgets changed since the current snapshot; run a full scan without `--repo`"
)]
PartialScanBudgetChanged,
/// An extractor artifact resolves outside its repository checkout.
#[error("artifact `{path}` resolves outside checkout `{checkout}`")]
ArtifactOutsideCheckout {
/// Configured artifact path.
path: PathBuf,
/// Canonical checkout root.
checkout: PathBuf,
},
/// Artifact path display contains unsafe metadata characters.
#[error("artifact path contains unsafe control or bidirectional characters")]
UnsafeArtifactPath,
/// Persistent storage failed.
#[error(transparent)]
Store(#[from] StoreError),
/// Federated graph construction or traversal failed.
#[error(transparent)]
Trace(#[from] TraceError),
/// Deterministic community analysis failed.
#[error(transparent)]
Community(#[from] CommunityError),
/// Ranked search or bounded traversal inputs were invalid.
#[error(transparent)]
Query(#[from] QueryError),
/// Federated impact or risk inputs were invalid.
#[error(transparent)]
Impact(#[from] ImpactError),
/// Git change collection or commit-gate input was invalid.
#[error(transparent)]
Change(#[from] ChangeError),
/// Semantic mapping or change-impact analysis failed.
#[error(transparent)]
ChangeAnalysis(#[from] ChangeAnalysisError),
/// Public interface request or rendering failed validation.
#[error(transparent)]
Interface(#[from] InterfaceError),
/// Opt-in pull-request inspection failed.
#[error(transparent)]
PullRequest(#[from] PullRequestError),
/// Workspace initialization could not be completed safely.
#[error("workspace initialization failed: {0}")]
Initialization(String),
/// A supervised worker observed a retryable operating-system or storage contention failure.
#[error("transient supervised execution failure: {0}")]
TransientExecution(String),
/// A worker preserved the public classification of an application failure across IPC.
#[error("supervised execution failed: {message}")]
SupervisedApplication {
/// Stable public exit classification produced inside the worker.
exit_code: ExitCode,
/// Bounded error explanation produced inside the worker.
message: String,
},
/// Client requested a traversal depth outside server bounds.
#[error("trace max_depth must be between 1 and {maximum}; received {found}")]
InvalidTraceDepth {
/// Requested depth.
found: usize,
/// Non-overridable server maximum.
maximum: usize,
},
}
/// Classifies application failures into stable process exit codes.
#[must_use]
pub const fn application_exit_code(error: &ApplicationError) -> ExitCode {
match error {
ApplicationError::SupervisedApplication { exit_code, .. } => *exit_code,
ApplicationError::Interface(source) => classify_interface_error(source),
ApplicationError::ExecutionLimit(code_system_graph_core::ExecutionLimitExceeded {
resource: code_system_graph_core::ExecutionResource::Cancellation,
..
})
| ApplicationError::Change(ChangeError::Cancelled)
| ApplicationError::PullRequest(PullRequestError::Cancelled) => ExitCode::Cancelled,
ApplicationError::Change(ChangeError::Timeout { .. })
| ApplicationError::PullRequest(PullRequestError::Timeout) => ExitCode::Timeout,
ApplicationError::WorkspaceNotFound(_) => ExitCode::NotFound,
ApplicationError::WorkspaceAlreadyExists(_) | ApplicationError::WatcherAlreadyActive(_) => {
ExitCode::Conflict
}
ApplicationError::Manifest(_)
| ApplicationError::UntrustedGlobalPolicySource { .. }
| ApplicationError::ExtractionLimit(_)
| ApplicationError::ExecutionLimit(_)
| ApplicationError::HttpExtraction(_)
| ApplicationError::PackageManifest(_)
| ApplicationError::GeneratedClient(_)
| ApplicationError::Graphql(_)
| ApplicationError::Event(_)
| ApplicationError::Protobuf(_)
| ApplicationError::Data(_)
| ApplicationError::Infrastructure(_)
| ApplicationError::Documentation(_)
| ApplicationError::ConfigExtraction(_)
| ApplicationError::SourceSyntax(_)
| ApplicationError::InvalidSourceObservation(_)
| ApplicationError::BatchPlan(_)
| ApplicationError::Link(_)
| ApplicationError::ManualLink(_)
| ApplicationError::Registry(_)
| ApplicationError::Config(_)
| ApplicationError::ManifestEdit(_)
| ApplicationError::UnknownOverrideRepository(_)
| ApplicationError::WorkspaceNameMismatch { .. }
| ApplicationError::ArtifactOutsideCheckout { .. }
| ApplicationError::UnsafeArtifactPath
| ApplicationError::PartialScanBudgetChanged
| ApplicationError::Trace(_)
| ApplicationError::Community(_)
| ApplicationError::Query(_)
| ApplicationError::Impact(_)
| ApplicationError::Change(_)
| ApplicationError::ChangeAnalysis(_)
| ApplicationError::PullRequest(_)
| ApplicationError::InvalidTraceDepth { .. } => ExitCode::InvalidInput,
ApplicationError::ReadFile { .. }
| ApplicationError::WriteFile { .. }
| ApplicationError::RegistryAliasMissing(_)
| ApplicationError::Store(_)
| ApplicationError::Initialization(_)
| ApplicationError::TransientExecution(_) => ExitCode::Internal,
}
}
pub(crate) fn load_execution_policy(
config_path: &Path,
) -> Result<ExecutionPolicy, ApplicationError> {
let manifest_source = read_file(config_path)?;
let manifest = parse_manifest(&manifest_source)?;
validate_global_policy_source(config_path, &manifest)?;
ExecutionPolicy::resolve(manifest.execution_policy.as_ref())
.map_err(ManifestError::from)
.map_err(Into::into)
}
/// Observable result of a successful scan.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ScanSummary {
/// Resource accounting for this supervised pass.
pub execution: code_system_graph_core::ExecutionSummary,
/// Workspace name.
pub workspace: String,
/// Published snapshot identifier.
pub snapshot_id: String,
/// Number of boundary nodes.
pub node_count: usize,
/// Number of confirmed cross-repository links.
pub edge_count: usize,
/// Number of direct evidence records.
pub evidence_count: usize,
/// Number of deterministic communities published with the graph.
pub community_count: usize,
/// Number of material community changes from the previous snapshot.
pub community_delta_count: usize,
/// Number of extractor-relevant files discovered.
pub discovered_input_count: usize,
/// Number of added, modified, or deleted inputs.
pub changed_input_count: usize,
/// Whether the current snapshot was reused without extractor execution.
pub reused_snapshot: bool,
/// Number of exact source symbols corroborated by optional local intelligence.
pub corroborated_symbol_count: usize,
/// Number of bounded affected-test paths reported by optional local intelligence.
pub affected_test_count: usize,
/// Total number of unique degradations before response truncation.
pub degradation_count: usize,
/// Explicit optional-provider degradation messages.
pub degradations: Vec<String>,
}
/// Highest-precedence command-line scan overrides.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ScanOverrides {
/// Optional requested workspace name, verified against the manifest before mutation.
pub workspace: Option<String>,
/// Per-repository `OpenAPI` artifact overrides keyed by alias.
pub repo_openapi: BTreeMap<String, String>,
/// Enable best-effort public `CodeGraph` corroboration.
pub codegraph: bool,
/// Optional explicit `CodeGraph` executable path.
pub codegraph_binary: Option<PathBuf>,
/// Restrict extractor work to one registered alias while reusing other repository batches.
pub repository: Option<String>,
/// Recompute selected extractor batches even when fingerprints are unchanged.
pub force: bool,
}
/// Action performed by one effective repository discovery rule.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum IgnoreRuleAction {
/// Omit matching paths from automatic discovery.
Exclude,
/// Re-enable paths otherwise matched by a built-in default exclusion.
Include,
}
/// Origin of one effective repository discovery rule.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum IgnoreRuleSource {
/// Non-overridable generated-state or version-control protection.
Protected,
/// Reactivable rule compiled into the binary.
BuiltInDefault,
/// Rule selected from the workspace manifest.
WorkspaceManifest,
/// Rule selected from repository-local configuration.
RepositoryLocal,
}
/// One ordered effective discovery rule returned by `config show`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct EffectiveIgnoreRule {
/// Rule action.
pub action: IgnoreRuleAction,
/// Rule origin.
pub source: IgnoreRuleSource,
/// Portable repository-relative glob.
pub pattern: String,
}
/// Configured pattern list and the precedence layer that selected it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ConfiguredPatterns {
/// Selected configuration layer.
pub source: ConfigSource,
/// Canonical configured patterns in deterministic order.
pub patterns: Vec<String>,
}
/// Complete observable ignore policy for one registered repository.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct IgnorePolicyReport {
/// Non-overridable exclusions compiled into the binary.
pub protected_excludes: Vec<String>,
/// Reactivable exclusions compiled into the binary.
pub default_excludes: Vec<String>,
/// Additional exclusions selected from configuration.
pub configured_excludes: ConfiguredPatterns,
/// Exceptions to built-in default exclusions selected from configuration.
pub include_defaults: ConfiguredPatterns,
/// Rules ordered from lowest to highest precedence.
pub effective_rules: Vec<EffectiveIgnoreRule>,
}
/// Effective configuration report for one repository alias.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct RepositoryConfigReport {
/// Workspace manifest alias.
pub alias: String,
/// Canonical checkout display path.
pub path: String,
/// Effective native automatic-discovery exclusions.
pub ignore_policy: IgnorePolicyReport,
}
/// Versioned read-only result returned by `csgraph config show`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ConfigReport {
/// Output schema version.
pub schema_version: u8,
/// Workspace name from the manifest.
pub workspace: String,
/// Effective global extraction safety limits, including applied defaults.
pub extraction_budgets: ExtractionBudgets,
/// Effective global supervised-execution policy, including applied defaults.
pub execution_policy: ExecutionPolicy,
/// Canonical fingerprint of the effective supervised-execution policy.
pub execution_policy_fingerprint: String,
/// Effective per-repository configuration in alias order.
pub repositories: Vec<RepositoryConfigReport>,
}
/// Current workspace registry and snapshot health.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct WorkspaceStatus {
/// Workspace name.
pub workspace: String,
/// Applied `SQLite` schema version.
pub schema_version: i64,
/// Result of the latest quick integrity check.
pub integrity_ok: bool,
/// Aggregate conservative freshness.
pub freshness: FreshnessSummary,
/// Per-checkout freshness in deterministic order.
pub repositories: Vec<RepoFreshness>,
/// Last persisted finite-watcher lifecycle state.
pub watcher: WatcherStatus,
}
/// Persisted lifecycle of the foreground sync watcher.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum WatcherState {
/// A verified watcher process currently owns the lease.
Active,
/// The watcher reached its inactivity deadline.
ExpiredIdle,
/// The watcher reached its absolute session deadline.
ExpiredSession,
/// A non-retryable execution limit ended the watcher.
FailedLimit,
/// The recorded process disappeared or no longer matches its start identity.
Stale,
/// No watcher session has been recorded for this workspace.
NeverStarted,
}
/// Observable finite-watcher status returned by `csgraph status`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct WatcherStatus {
/// Stable lifecycle state.
pub state: WatcherState,
/// Bounded termination or stale-state diagnostic.
pub detail: Option<String>,
}
/// Result of an explicit database backup.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct BackupSummary {
/// Source database display path.
pub database: String,
/// Created backup display path.
pub backup: String,
}
/// Result of an explicit database restore.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct RestoreSummary {
/// Selected backup source display path.
pub source: String,
/// Safety backup of the replaced database.
pub safety_backup: Option<String>,
/// Schema version after restore.
pub schema_version: i64,
}
/// Compact workspace registry item returned by CLI listing.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct WorkspaceRegistryItem {
/// Stable workspace identifier.
pub id: String,
/// User-facing workspace name.
pub name: String,
/// Current merged manifest fingerprint.
pub manifest_hash: String,
/// Canonical manifest display path.
pub config_path: Option<String>,
/// Number of registered repository aliases.
pub repository_count: usize,
}
/// Preview or result of a constrained repository manifest mutation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ManifestMutationSummary {
/// Human-readable operation summary.
pub summary: String,
/// Manifest display path.
pub manifest: String,
/// Backup created before an applied mutation.
pub backup: Option<String>,
/// Whether the mutation was committed.
pub applied: bool,
/// Complete proposed manifest for dry-run review.
pub rendered_manifest: Option<String>,
}
/// Result of adding or removing a persisted workspace registration.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct WorkspaceMutationSummary {
/// Workspace name.
pub workspace: String,
/// Canonical manifest display path when adding.
pub config_path: Option<String>,
/// Number of registered repository aliases before removal or after addition.
pub repository_count: usize,
/// Mutation kind.
pub operation: String,
}
/// Input contract shared by CLI and MCP trace delivery.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct TraceInput {
/// Stable source node identifier.
pub from: String,
/// Stable target node identifier.
pub to: String,
/// Maximum number of traversed edges.
#[serde(default = "default_max_depth")]
#[schemars(range(min = 1, max = 32))]
pub max_depth: usize,
}
/// Safe ranked-search input shared by CLI and MCP delivery.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct SearchInput {
/// Text matched against graph identities and the current FTS index.
pub query: String,
/// Optional entity-kind filter.
#[serde(default)]
pub node_kinds: Vec<NodeKind>,
/// Optional repository filter.
#[serde(default)]
pub repo_ids: Vec<RepoId>,
/// Optional exact service node scopes.
#[serde(default)]
pub service_ids: Vec<NodeId>,
/// Optional exact community scopes.
#[serde(default)]
pub community_ids: Vec<CommunityId>,
/// Zero-based result offset.
#[serde(default)]
pub offset: usize,
/// Bounded page size.
#[serde(default = "default_search_limit")]
#[schemars(range(min = 1, max = 100))]
pub limit: usize,
}
/// Read-only request for one deterministic community page or snapshot comparison.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct CommunityInput {
/// Optional exact community identity.
pub community_id: Option<CommunityId>,
/// Optional historical snapshot to compare against the current snapshot.
pub compare_snapshot_id: Option<String>,
/// Zero-based community offset.
#[serde(default)]
pub offset: usize,
/// Bounded page size.
#[serde(default = "default_community_limit")]
#[schemars(range(min = 1, max = 100))]
pub limit: usize,
}
/// Paginated and optionally comparative community result.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct CommunityReport {
/// Current graph snapshot identity.
pub snapshot_id: String,
/// Community engine semantic version.
pub engine_version: String,
/// Reproducibility configuration.
pub config: CommunityConfig,
/// Selected communities in deterministic relevance order.
pub communities: Vec<Community>,
/// Total communities before pagination.
pub total_communities: usize,
/// Applied zero-based offset.
pub offset: usize,
/// Applied page size.
pub limit: usize,
/// Whether more communities remain.
pub truncated: bool,
/// Optional material delta from the requested historical snapshot.
pub delta: Option<CommunityDelta>,
}
/// Read-only local change collection input shared by CLI and MCP.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ChangesInput {
/// Registered repository alias.
pub repository: String,
/// Local Git state or committed comparison to inspect.
pub scope: ChangeScope,
}
/// Ephemeral intra-repository source and flow exploration input.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ExploreInput {
/// Workspace selected by the delivery adapter.
pub workspace: String,
/// Registered repository alias; omitted only when the workspace contains one repository.
#[serde(default)]
pub repository: Option<String>,
/// Focused symbol, flow, architecture, or implementation question.
pub query: String,
/// Maximum source files returned by the local provider.
#[serde(default = "default_explore_max_files")]
#[schemars(range(min = 1, max = 25))]
pub max_files: usize,
}
const fn default_explore_max_files() -> usize {
12
}
/// Explicit remote pull-request inspection input shared by CLI and MCP.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct PullRequestInput {
/// Public provider contract.
pub provider: PullRequestProviderKind,
/// GitHub owner or Bitbucket workspace.
pub owner: String,
/// GitHub repository or Bitbucket repository slug.
pub repository: String,
/// Provider-native pull-request number.
pub number: u64,
/// Per-request confirmation that remote access is permitted.
pub consent_to_remote_access: bool,
}
/// Explicit hosted pull-request list input shared by delivery adapters.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct PullRequestListInput {
/// Public provider contract.
pub provider: PullRequestProviderKind,
/// GitHub owner or Bitbucket workspace.
pub owner: String,
/// GitHub repository or Bitbucket repository slug.
pub repository: String,
/// Provider-native lifecycle filter.
pub state: PullRequestListState,
/// Opaque positive page cursor returned by a prior response.
pub cursor: Option<String>,
/// Maximum number of source-free summaries.
pub limit: usize,
/// Per-request confirmation that remote access is permitted.
pub consent_to_remote_access: bool,
}
/// Result of creating a new strict workspace manifest.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct InitReport {
/// Created manifest path.
pub manifest_path: PathBuf,
/// Workspace-local Git ignore file when the workspace belongs to a Git worktree.
pub gitignore_path: Option<PathBuf>,
/// Whether initialization added the generated-state rule.
pub gitignore_updated: bool,
/// Validated workspace name.
pub workspace: String,
/// Manifest contract version.
pub manifest_version: u32,
}
/// Explicit source-free diagnostic bundle suitable for local support handoff.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct DiagnosticBundle {
/// Public bundle schema version.
pub schema_version: u32,
/// `Code System Graph` binary version.
pub binary_version: String,
/// Operating-system target reported by the running binary.
pub operating_system: String,
/// CPU architecture reported by the running binary.
pub architecture: String,
/// Bundle generation time as Unix milliseconds.
pub generated_at_unix_ms: u128,
/// Whether extended local diagnostics were explicitly requested through the environment.
pub debug_requested: bool,
/// Conservative source-free doctor report.
pub doctor: DoctorReport,
}
struct WorkspaceContext {
manifest: WorkspaceManifest,
registry: RegisteredWorkspace,
repository_configs: BTreeMap<String, EffectiveRepositoryConfig>,
extraction_budgets: ExtractionBudgets,
execution_policy: ExecutionPolicy,
}
/// Resolves and reports native discovery exclusions without opening a graph database.
///
/// # Errors
///
/// Returns [`ApplicationError`] when the manifest, repository-local configuration, checkout, or
/// optional repository selection is invalid.
pub fn show_config(
config_path: &Path,
selected_repository: Option<&str>,
) -> Result<ConfigReport, ApplicationError> {
let context = load_workspace_context(config_path, &ScanOverrides::default())?;
if let Some(selected) = selected_repository
&& !context.manifest.repos.contains_key(selected)
{
return Err(ApplicationError::UnknownOverrideRepository(
selected.to_owned(),
));
}
let repositories = context
.manifest
.repos
.keys()
.filter(|alias| selected_repository.is_none_or(|selected| selected == alias.as_str()))
.map(|alias| {
let path = context
.registry
.checkout_path(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let effective = context
.repository_configs
.get(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
Ok(RepositoryConfigReport {
alias: alias.clone(),
path: path.display().to_string(),
ignore_policy: ignore_policy_report(&effective.ignore_policy),
})
})
.collect::<Result<Vec<_>, ApplicationError>>()?;
Ok(ConfigReport {
schema_version: 1,
workspace: context.manifest.name,
extraction_budgets: context.extraction_budgets,
execution_policy_fingerprint: context.execution_policy.fingerprint(),
execution_policy: context.execution_policy,
repositories,
})
}
fn ignore_policy_report(policy: &IgnorePolicy) -> IgnorePolicyReport {
let configured_source = rule_source(policy.configured_excludes_source());
let include_source = rule_source(policy.include_defaults_source());
let mut effective_rules = DEFAULT_EXCLUDES
.iter()
.map(|pattern| EffectiveIgnoreRule {
action: IgnoreRuleAction::Exclude,
source: IgnoreRuleSource::BuiltInDefault,
pattern: (*pattern).to_owned(),
})
.chain(
policy
.include_defaults()
.iter()
.map(|pattern| EffectiveIgnoreRule {
action: IgnoreRuleAction::Include,
source: include_source,
pattern: pattern.clone(),
}),
)
.chain(
policy
.configured_excludes()
.iter()
.map(|pattern| EffectiveIgnoreRule {
action: IgnoreRuleAction::Exclude,
source: configured_source,
pattern: pattern.clone(),
}),
)
.collect::<Vec<_>>();
effective_rules.extend(
PROTECTED_EXCLUDES
.iter()
.map(|pattern| EffectiveIgnoreRule {
action: IgnoreRuleAction::Exclude,
source: IgnoreRuleSource::Protected,
pattern: (*pattern).to_owned(),
}),
);
IgnorePolicyReport {
protected_excludes: PROTECTED_EXCLUDES
.iter()
.map(|pattern| (*pattern).to_owned())
.collect(),
default_excludes: DEFAULT_EXCLUDES
.iter()
.map(|pattern| (*pattern).to_owned())
.collect(),
configured_excludes: ConfiguredPatterns {
source: policy.configured_excludes_source(),
patterns: policy.configured_excludes().to_vec(),
},
include_defaults: ConfiguredPatterns {
source: policy.include_defaults_source(),
patterns: policy.include_defaults().to_vec(),
},
effective_rules,
}
}
const fn rule_source(source: ConfigSource) -> IgnoreRuleSource {
match source {
ConfigSource::WorkspaceManifest => IgnoreRuleSource::WorkspaceManifest,
ConfigSource::RepositoryLocal => IgnoreRuleSource::RepositoryLocal,
ConfigSource::CliOverride | ConfigSource::AutoDetected | ConfigSource::Default => {
IgnoreRuleSource::BuiltInDefault
}
}
}
struct GraphAssembly {
nodes: Vec<Node>,
edges: Vec<Edge>,
evidence: Vec<Evidence>,
link_decisions: Vec<LinkDecision>,
link_node_keys: BTreeMap<NodeId, String>,
degradations: Vec<String>,
}
struct FocusedBatchState {
source_batches: Vec<ExtractorBatch<SourceObservation>>,
previous_source_batches: Vec<ExtractorBatch<SourceObservation>>,
package_batches: Vec<ExtractorBatch<PackageManifest>>,
generated_client_batches: Vec<ExtractorBatch<GeneratedClientMetadata>>,
graphql_batches: Vec<ExtractorBatch<GraphqlDocument>>,
event_batches: Vec<ExtractorBatch<EventDocument>>,
protobuf_batches: Vec<ExtractorBatch<ProtobufDocument>>,
data_batches: Vec<ExtractorBatch<DataDocument>>,
infrastructure_batches: Vec<ExtractorBatch<InfrastructureDocument>>,
documentation_batches: Vec<ExtractorBatch<DocumentationDocument>>,
config_batches: Vec<ExtractorBatch<SafeConfigDocument>>,
stored_batches: Vec<StoredExtractorBatch>,
degradations: Vec<String>,
force_relink: bool,
checkpoint_writes: u64,
artifact_durations_ms: Vec<u64>,
}
struct RepositoryCorroboration {
repo_id: RepoId,
report: CorroborationReport,
}
#[derive(Default)]
struct CorroborationSummary {
reports: Vec<RepositoryCorroboration>,
confirmed_symbols: usize,
affected_tests: usize,
degradations: Vec<String>,
}
fn default_max_depth() -> usize {
8
}
fn current_unix_millis() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.and_then(|duration| u64::try_from(duration.as_millis()).ok())
.unwrap_or(u64::MAX)
}
fn duration_millis(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
fn artifact_execution_summary(durations_ms: &[u64]) -> code_system_graph_core::ExecutionSummary {
let mut ordered = durations_ms.to_vec();
ordered.sort_unstable();
code_system_graph_core::ExecutionSummary {
measured_artifacts: u64::try_from(ordered.len()).unwrap_or(u64::MAX),
artifact_duration_p50_ms: duration_percentile(&ordered, 50),
artifact_duration_p95_ms: duration_percentile(&ordered, 95),
artifact_duration_p99_ms: duration_percentile(&ordered, 99),
..code_system_graph_core::ExecutionSummary::default()
}
}
fn duration_percentile(ordered: &[u64], percentile: usize) -> u64 {
if ordered.is_empty() {
return 0;
}
let index = ordered
.len()
.saturating_mul(percentile)
.div_ceil(100)
.saturating_sub(1)
.min(ordered.len() - 1);
ordered[index]
}
fn default_search_limit() -> usize {
20
}
fn default_community_limit() -> usize {
20
}
fn default_community_config() -> CommunityConfig {
CommunityConfig {
algorithm: CommunityAlgorithm::Louvain,
scope: CommunityScope::Federated,
seed: 0,
resolution: 1.0,
minimum_confidence: 0.5,
edge_weights: Vec::new(),
max_iterations: 100,
}
}
/// Creates a strict minimal workspace manifest without overwriting existing files.
///
/// # Errors
///
/// When the workspace belongs to a Git worktree, generated local state is excluded through an
/// idempotent `.gitignore` update that preserves all existing bytes and rules. No `.gitignore` is
/// created for an unversioned directory that only contains child repositories.
///
/// Returns [`ApplicationError`] when the directory cannot be created, the destination already
/// exists, the derived name is invalid, or the manifest and ignore rule cannot be durably written.
pub fn initialize_workspace(
directory: &Path,
requested_name: Option<&str>,
) -> Result<InitReport, ApplicationError> {
fs::create_dir_all(directory).map_err(|error| {
ApplicationError::Initialization(format!(
"cannot create `{}`: {error}",
directory.display()
))
})?;
let workspace = requested_name.map_or_else(
|| {
directory
.file_name()
.and_then(OsStr::to_str)
.filter(|name| !name.is_empty())
.unwrap_or("workspace")
.to_owned()
},
str::to_owned,
);
let encoded_name = serde_json::to_string(&workspace)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
let source = format!("version: 1\nname: {encoded_name}\nrepos:\n root:\n path: .\n");
let manifest = parse_manifest(&source)?;
let manifest_path = directory.join("code-system-graph.yaml");
let mut destination = fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&manifest_path)
.map_err(|error| {
ApplicationError::Initialization(format!(
"cannot create `{}` without overwriting: {error}",
manifest_path.display()
))
})?;
destination.write_all(source.as_bytes()).map_err(|error| {
ApplicationError::Initialization(format!(
"cannot write `{}`: {error}",
manifest_path.display()
))
})?;
destination.sync_all().map_err(|error| {
ApplicationError::Initialization(format!(
"cannot synchronize `{}`: {error}",
manifest_path.display()
))
})?;
drop(destination);
let (gitignore_path, gitignore_updated) = configure_generated_state_ignore(directory)
.inspect_err(|_| {
let _ = fs::remove_file(&manifest_path);
})?;
Ok(InitReport {
manifest_path,
gitignore_path,
gitignore_updated,
workspace: manifest.name,
manifest_version: manifest.version,
})
}
fn configure_generated_state_ignore(
directory: &Path,
) -> Result<(Option<PathBuf>, bool), ApplicationError> {
let canonical_directory = fs::canonicalize(directory).map_err(|error| {
ApplicationError::Initialization(format!(
"cannot resolve workspace directory `{}`: {error}",
directory.display()
))
})?;
if !belongs_to_git_worktree(&canonical_directory) {
return Ok((None, false));
}
let (path, updated) = ensure_generated_state_ignored(directory)?;
Ok((Some(path), updated))
}
fn belongs_to_git_worktree(directory: &Path) -> bool {
directory
.ancestors()
.any(|ancestor| valid_git_worktree_marker(&ancestor.join(".git")))
}
fn valid_git_worktree_marker(marker: &Path) -> bool {
let Ok(metadata) = fs::symlink_metadata(marker) else {
return false;
};
if metadata.file_type().is_dir() {
return fs::symlink_metadata(marker.join("HEAD"))
.is_ok_and(|head| head.file_type().is_file());
}
if !metadata.file_type().is_file() || metadata.len() > 4_096 {
return false;
}
let Ok(source) = fs::read_to_string(marker) else {
return false;
};
source
.lines()
.next()
.is_some_and(|line| line.trim_start().starts_with("gitdir:"))
}
fn ensure_generated_state_ignored(directory: &Path) -> Result<(PathBuf, bool), ApplicationError> {
let gitignore_path = directory.join(".gitignore");
let mut content = match fs::read(&gitignore_path) {
Ok(content) => content,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(error) => {
return Err(ApplicationError::Initialization(format!(
"cannot read `{}` before updating generated-state rules: {error}",
gitignore_path.display()
)));
}
};
if generated_state_is_ignored(&content) {
return Ok((gitignore_path, false));
}
if !content.is_empty() && !content.ends_with(b"\n") {
content.push(b'\n');
}
content.extend_from_slice(GENERATED_STATE_IGNORE_RULE);
content.push(b'\n');
let mut destination = AtomicWriteFile::open(&gitignore_path).map_err(|error| {
ApplicationError::Initialization(format!(
"cannot prepare `{}` for generated-state rules: {error}",
gitignore_path.display()
))
})?;
destination
.write_all(&content)
.and_then(|()| destination.sync_all())
.map_err(|error| {
ApplicationError::Initialization(format!(
"cannot write `{}`: {error}",
gitignore_path.display()
))
})?;
destination.commit().map_err(|error| {
ApplicationError::Initialization(format!(
"cannot commit `{}`: {error}",
gitignore_path.display()
))
})?;
Ok((gitignore_path, true))
}
fn generated_state_is_ignored(content: &[u8]) -> bool {
content
.split(|byte| *byte == b'\n')
.map(|line| line.strip_suffix(b"\r").map_or(line, |trimmed| trimmed))
.fold(None, |state, line| match line {
b".code-system-graph/"
| b"/.code-system-graph/"
| b".code-system-graph"
| b"/.code-system-graph" => Some(true),
b"!.code-system-graph/"
| b"!/.code-system-graph/"
| b"!.code-system-graph"
| b"!/.code-system-graph" => Some(false),
_ => state,
})
.unwrap_or(false)
}
/// Scans the initial HTTP vertical slice and publishes one atomic snapshot.
///
/// # Errors
///
/// Returns [`ApplicationError`] for unreadable inputs, invalid contracts, invalid manual links, or
/// storage failure. Ambiguous automatic HTTP providers are reported as degradations.
pub fn scan_workspace(
config_path: &Path,
database_path: &Path,
) -> Result<ScanSummary, ApplicationError> {
scan_workspace_with_overrides(config_path, database_path, &ScanOverrides::default())
}
/// Scans with explicit highest-precedence command-line overrides.
///
/// # Errors
///
/// Returns [`ApplicationError`] for unknown aliases, invalid overrides, unreadable inputs,
/// invalid contracts, invalid manual links, or storage failure. Ambiguous automatic HTTP providers
/// are reported as degradations.
pub fn scan_workspace_with_overrides(
config_path: &Path,
database_path: &Path,
overrides: &ScanOverrides,
) -> Result<ScanSummary, ApplicationError> {
worker::supervise_scan(config_path, database_path, overrides)
}
/// Scans through an explicitly selected compatible worker executable.
///
/// This entry point lets an embedding application re-execute its own binary after dispatching
/// `__worker-v1` to [`run_worker_from_stdio`], so using the library does not require Cargo to have
/// built the `csgraph` binary target next to the host executable.
///
/// # Errors
///
/// Returns [`ApplicationError`] when the worker cannot start or the supervised scan fails.
pub fn scan_workspace_with_worker_executable(
config_path: &Path,
database_path: &Path,
overrides: &ScanOverrides,
worker_executable: &Path,
) -> Result<ScanSummary, ApplicationError> {
worker::supervise_scan_with_executable(config_path, database_path, overrides, worker_executable)
}
#[doc(hidden)]
pub(crate) fn scan_workspace_direct(
config_path: &Path,
database_path: &Path,
overrides: &ScanOverrides,
) -> Result<ScanSummary, ApplicationError> {
scan_workspace_direct_with_mode(config_path, database_path, overrides, false)
}
pub(crate) fn scan_workspace_direct_for_sync(
config_path: &Path,
database_path: &Path,
overrides: &ScanOverrides,
reuse_unchanged_with_codegraph: bool,
) -> Result<ScanSummary, ApplicationError> {
scan_workspace_direct_with_mode(
config_path,
database_path,
overrides,
reuse_unchanged_with_codegraph,
)
}
#[expect(
clippy::too_many_lines,
reason = "Atomic scan orchestration keeps lock, resume, publication, and summary sequencing visible"
)]
fn scan_workspace_direct_with_mode(
config_path: &Path,
database_path: &Path,
overrides: &ScanOverrides,
reuse_unchanged_with_codegraph: bool,
) -> Result<ScanSummary, ApplicationError> {
let context = load_workspace_context(config_path, overrides)?;
worker::report_progress(code_system_graph_core::JobPhase::Configuration, 1);
if let Some(requested) = &overrides.workspace
&& requested != &context.manifest.name
{
return Err(ApplicationError::WorkspaceNameMismatch {
requested: requested.clone(),
manifest: context.manifest.name,
});
}
let mut fingerprints = discover_artifact_fingerprints(&context)?;
worker::report_progress(
code_system_graph_core::JobPhase::Fingerprinting,
u64::try_from(fingerprints.len()).unwrap_or(u64::MAX),
);
let _writer_lock = StoreLock::acquire(database_path, Duration::from_mins(5))?;
let mut store = SqliteStore::open(database_path)?;
let database_instance_id = store.database_instance_id()?;
let mut work_state = work_state::WorkState::open(database_path, &database_instance_id)
.map_err(ApplicationError::Initialization)?;
let mut previous_fingerprints =
match store.load_current_artifact_fingerprints(&context.manifest.name) {
Ok(previous) => previous,
Err(StoreError::CurrentSnapshotMissing(_)) => Vec::new(),
Err(error) => return Err(error.into()),
};
if let Some(alias) = &overrides.repository {
let selected = context
.registry
.record
.repositories
.iter()
.find(|repository| &repository.alias == alias)
.ok_or_else(|| ApplicationError::UnknownOverrideRepository(alias.clone()))?
.id
.clone();
fingerprints.retain(|fingerprint| fingerprint.repo_id == selected);
fingerprints.extend(
previous_fingerprints
.iter()
.filter(|fingerprint| fingerprint.repo_id != selected)
.cloned(),
);
fingerprints.sort_by_key(|fingerprint| ArtifactKey::from(fingerprint));
if overrides.force {
previous_fingerprints.retain(|fingerprint| fingerprint.repo_id != selected);
}
} else if overrides.force {
previous_fingerprints.clear();
}
let fingerprint_hashes = fingerprints
.iter()
.map(|fingerprint| fingerprint.content_hash.as_str())
.collect::<Vec<_>>()
.join(":");
let snapshot_id = stable_id(
"snapshot",
&format!(
"{}:{fingerprint_hashes}:community-engine-v1",
context.registry.record.manifest_hash
),
);
let mut previous_extractor_batches = match store.load_current_extractor_batches_with_limit(
&context.manifest.name,
context
.extraction_budgets
.max_serialized_output_bytes_per_artifact,
) {
Ok(previous) => previous,
Err(StoreError::CurrentSnapshotMissing(_)) => Vec::new(),
Err(error) => return Err(error.into()),
};
let budget_fingerprint = context.extraction_budgets.fingerprint();
let candidate_material = serde_json::to_vec(&(
&context.registry.record.manifest_hash,
&fingerprints,
&budget_fingerprint,
EXTRACTION_CONTRACT_VERSION,
))
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
let candidate_fingerprint = stable_id_bytes("work-candidate-v1", &candidate_material);
let _resumed_candidate = work_state
.begin_candidate(
&context.manifest.name,
&candidate_fingerprint,
current_unix_millis(),
)
.map_err(ApplicationError::Initialization)?;
let cached_batches = if overrides.force {
Vec::new()
} else {
work_state
.load_batches(
&fingerprints,
&budget_fingerprint,
EXTRACTION_CONTRACT_VERSION,
context
.extraction_budgets
.max_serialized_output_bytes_per_artifact,
current_unix_millis(),
)
.map_err(ApplicationError::Initialization)?
};
let cached_keys = cached_batches
.iter()
.map(|batch| ArtifactKey::from(&batch.source))
.collect::<BTreeSet<_>>();
for cached in cached_batches {
let key = ArtifactKey::from(&cached.source);
if !previous_extractor_batches
.iter()
.any(|batch| ArtifactKey::from(&batch.source) == key && batch.source == cached.source)
{
previous_extractor_batches.push(cached);
}
}
let checkpoint_hits = u64::try_from(cached_keys.len()).unwrap_or(u64::MAX);
if overrides.repository.is_some()
&& previous_extractor_batches
.iter()
.any(|batch| batch.budget_fingerprint != budget_fingerprint)
{
return Err(ApplicationError::PartialScanBudgetChanged);
}
let previous_graph = match store.load_current_graph(&context.manifest.name) {
Ok(graph) => graph,
Err(StoreError::CurrentSnapshotMissing(_)) => (Vec::new(), Vec::new()),
Err(error) => return Err(error.into()),
};
let previous_communities = match store.load_current_community_snapshot(&context.manifest.name) {
Ok(snapshot) => Some(snapshot),
Err(StoreError::CurrentSnapshotMissing(_) | StoreError::CommunitySnapshotMissing(_)) => {
None
}
Err(error) => return Err(error.into()),
};
let previous_manifest_matches = match store.load_workspace_registry(&context.manifest.name) {
Ok(previous) => previous.manifest_hash == context.registry.record.manifest_hash,
Err(StoreError::RegistryIncomplete(_)) => false,
Err(error) => return Err(error.into()),
};
let plan = plan_incremental_scan(&previous_fingerprints, &fingerprints);
let staged_snapshot = if let Ok(candidate) =
work_state.load_candidate_snapshot(&context.manifest.name, &candidate_fingerprint)
{
candidate
} else {
work_state
.complete_candidate(&context.manifest.name)
.map_err(ApplicationError::Initialization)?;
work_state
.begin_candidate(
&context.manifest.name,
&candidate_fingerprint,
current_unix_millis(),
)
.map_err(ApplicationError::Initialization)?;
None
};
if let Some(candidate) = staged_snapshot {
publish_snapshot_candidate(
&mut store,
SnapshotBatch {
workspace: &context.registry.record,
snapshot_id: &candidate.snapshot_id,
nodes: &candidate.nodes,
edges: &candidate.edges,
evidence: &candidate.evidence,
fingerprints: &candidate.fingerprints,
extractor_batches: &candidate.extractor_batches,
extractor_runs: &candidate.extractor_runs,
manual_links: &candidate.manual_links,
community_snapshot: Some(&candidate.community_snapshot),
},
)?;
work_state
.complete_candidate(&context.manifest.name)
.map_err(ApplicationError::Initialization)?;
let mut execution = candidate.execution;
execution.checkpoint_hits = checkpoint_hits;
return Ok(ScanSummary {
execution,
workspace: context.manifest.name,
snapshot_id: candidate.snapshot_id,
node_count: candidate.nodes.len(),
edge_count: candidate.edges.len(),
evidence_count: candidate.evidence.len(),
community_count: candidate.community_snapshot.communities.len(),
community_delta_count: candidate.community_delta_count,
discovered_input_count: candidate.fingerprints.len(),
changed_input_count: plan.changed_count(),
reused_snapshot: false,
corroborated_symbol_count: candidate.corroborated_symbol_count,
affected_test_count: candidate.affected_test_count,
degradation_count: candidate.degradations.len(),
degradations: candidate.degradations,
});
}
if previous_manifest_matches
&& !plan.has_changes()
&& focused_batch_cache_complete(
&fingerprints,
&previous_extractor_batches,
&context.extraction_budgets,
)
&& previous_communities.is_some()
&& (!overrides.codegraph || reuse_unchanged_with_codegraph)
{
work_state
.complete_candidate(&context.manifest.name)
.map_err(ApplicationError::Initialization)?;
let current = store.current_snapshot_summary(&context.manifest.name)?;
let (degradation_count, degradations) = finalize_scan_degradations(
stored_batch_degradations(&previous_extractor_batches, &context.extraction_budgets)?,
);
return Ok(ScanSummary {
execution: code_system_graph_core::ExecutionSummary {
checkpoint_hits,
..code_system_graph_core::ExecutionSummary::default()
},
workspace: context.manifest.name,
snapshot_id: current.snapshot_id,
node_count: current.node_count,
edge_count: current.edge_count,
evidence_count: current.evidence_count,
community_count: previous_communities
.as_ref()
.map_or(0, |snapshot| snapshot.communities.len()),
community_delta_count: 0,
discovered_input_count: fingerprints.len(),
changed_input_count: 0,
reused_snapshot: true,
corroborated_symbol_count: 0,
affected_test_count: 0,
degradation_count,
degradations,
});
}
let batch_plan = plan_extractor_batches(&plan);
let focused_batches = assemble_focused_batches(
&context,
&fingerprints,
&previous_extractor_batches,
&batch_plan,
&cached_keys,
&mut work_state,
)?;
work_state
.set_candidate_phase(&context.manifest.name, "extracted", current_unix_millis())
.map_err(ApplicationError::Initialization)?;
worker::report_progress(
code_system_graph_core::JobPhase::Extraction,
u64::try_from(focused_batches.stored_batches.len()).unwrap_or(u64::MAX),
);
let mut graph = assemble_graph(&context, &fingerprints, &focused_batches)?;
relink_affected_graph(
&mut graph,
&plan,
&batch_plan,
&focused_batches,
&previous_graph.0,
&previous_graph.1,
)?;
let corroboration = if overrides.codegraph {
run_codegraph_corroboration(
&context,
&focused_batches,
&plan,
overrides.repository.as_deref(),
overrides.codegraph_binary.clone(),
)
} else {
CorroborationSummary::default()
};
apply_codegraph_corroboration(&mut graph, &corroboration.reports);
let manual_links =
resolve_manual_links(&context.manifest.manual_links, &graph.nodes, &graph.edges)?;
graph.edges = manual_links.edges;
graph.evidence.extend(manual_links.evidence);
graph.evidence.sort_by(|left, right| left.id.cmp(&right.id));
graph.evidence.dedup_by(|left, right| left.id == right.id);
graph.link_decisions = manual_links.decisions;
worker::report_progress(
code_system_graph_core::JobPhase::GraphAssembly,
u64::try_from(graph.nodes.len().saturating_add(graph.edges.len())).unwrap_or(u64::MAX),
);
let GraphAssembly {
nodes,
edges,
evidence,
link_decisions,
degradations: graph_degradations,
..
} = graph;
let community_config = default_community_config();
let community_snapshot = previous_communities
.as_ref()
.filter(|previous| {
previous.snapshot_id == snapshot_id
&& previous.config == community_config
&& community_topology_unchanged(
&previous_graph.0,
&previous_graph.1,
&nodes,
&edges,
)
})
.cloned()
.map_or_else(
|| {
analyze_communities_with_progress(
&snapshot_id,
&nodes,
&edges,
community_config.clone(),
|completed| {
worker::report_progress(
code_system_graph_core::JobPhase::Communities,
completed,
);
},
)
},
Ok,
)?;
drop(previous_graph);
let community_delta_count = previous_communities.as_ref().map_or(0, |previous| {
compare_community_snapshots(previous, &community_snapshot)
.changes
.len()
});
worker::report_progress(
code_system_graph_core::JobPhase::Communities,
u64::try_from(community_snapshot.communities.len()).unwrap_or(u64::MAX),
);
let extractor_runs = extractor_runs(&snapshot_id, &fingerprints, &plan);
let manual_link_records = persisted_manual_link_records(&snapshot_id, &link_decisions)?;
let mut staged_degradations = focused_batches.degradations.clone();
staged_degradations.extend(graph_degradations);
staged_degradations.extend(corroboration.degradations.clone());
staged_degradations.extend(stored_batch_degradations(
&focused_batches.stored_batches,
&context.extraction_budgets,
)?);
let (_, staged_degradations) = finalize_scan_degradations(staged_degradations);
let mut artifact_execution = artifact_execution_summary(&focused_batches.artifact_durations_ms);
artifact_execution.checkpoint_hits = checkpoint_hits;
artifact_execution.checkpoints_written = focused_batches.checkpoint_writes;
let candidate = work_state::StagedSnapshot {
snapshot_id,
nodes,
edges,
evidence,
fingerprints,
extractor_batches: focused_batches.stored_batches,
extractor_runs,
manual_links: manual_link_records,
community_snapshot,
community_delta_count,
corroborated_symbol_count: corroboration.confirmed_symbols,
affected_test_count: corroboration.affected_tests,
execution: artifact_execution,
degradations: staged_degradations,
};
work_state
.store_candidate_snapshot(
&context.manifest.name,
&candidate_fingerprint,
&candidate,
current_unix_millis(),
)
.map_err(ApplicationError::Initialization)?;
publish_snapshot_candidate(
&mut store,
SnapshotBatch {
workspace: &context.registry.record,
snapshot_id: &candidate.snapshot_id,
nodes: &candidate.nodes,
edges: &candidate.edges,
evidence: &candidate.evidence,
fingerprints: &candidate.fingerprints,
extractor_batches: &candidate.extractor_batches,
extractor_runs: &candidate.extractor_runs,
manual_links: &candidate.manual_links,
community_snapshot: Some(&candidate.community_snapshot),
},
)?;
let mut degradations = candidate.degradations.clone();
for item in &corroboration.reports {
let Some(capability) = &item.report.capability else {
continue;
};
let record = provider_capability_record(&context.manifest.name, &item.repo_id, capability);
if let Err(error) = store.upsert_provider_capabilities(&record) {
degradations.push(format!(
"CodeGraph capability persistence for `{}` degraded: {error}",
item.repo_id.as_str()
));
}
}
let (degradation_count, degradations) = finalize_scan_degradations(degradations);
let _ = work_state.complete_candidate(&context.manifest.name);
Ok(ScanSummary {
execution: candidate.execution,
workspace: context.manifest.name,
snapshot_id: candidate.snapshot_id,
node_count: candidate.nodes.len(),
edge_count: candidate.edges.len(),
evidence_count: candidate.evidence.len(),
community_count: candidate.community_snapshot.communities.len(),
community_delta_count: candidate.community_delta_count,
discovered_input_count: candidate.fingerprints.len(),
changed_input_count: plan.changed_count(),
reused_snapshot: false,
corroborated_symbol_count: candidate.corroborated_symbol_count,
affected_test_count: candidate.affected_test_count,
degradation_count,
degradations,
})
}
fn publish_snapshot_candidate(
store: &mut SqliteStore,
batch: SnapshotBatch<'_>,
) -> Result<(), StoreError> {
const REPORT_INTERVAL: u64 = 1_024;
let mut pending = 0_u64;
worker::report_progress(code_system_graph_core::JobPhase::Publication, 1);
store.publish_snapshot_with_progress(batch, |rows| {
worker::check_time(code_system_graph_core::JobPhase::Publication);
pending = pending.saturating_add(rows);
if pending >= REPORT_INTERVAL {
worker::report_progress(code_system_graph_core::JobPhase::Publication, pending);
pending = 0;
}
})?;
Ok(())
}
/// Traces the current stored graph and returns a versioned conservative envelope.
///
/// # Errors
///
/// Returns [`ApplicationError`] for storage, graph validation, or invalid anchor failures.
pub fn trace_workspace(
database_path: &Path,
workspace: &str,
input: &TraceInput,
) -> Result<ToolEnvelope<TraceReport>, ApplicationError> {
if !(1..=MAX_TRACE_DEPTH).contains(&input.max_depth) {
return Err(ApplicationError::InvalidTraceDepth {
found: input.max_depth,
maximum: MAX_TRACE_DEPTH,
});
}
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let freshness = freshness_summary(&store.load_current_freshness(workspace)?);
let graph = FederatedGraph::new(nodes, edges)?;
let report = graph.trace(
&NodeId::new(&input.from),
&NodeId::new(&input.to),
input.max_depth,
)?;
let status = if report.coverage_gaps.is_empty() && freshness.overall == OverallFreshness::Fresh
{
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness,
warnings: Vec::new(),
})
}
/// Searches current graph entities with structural, FTS, community, evidence, and freshness signals.
///
/// # Errors
///
/// Returns [`ApplicationError`] for invalid input, missing snapshots, malformed persistence, or
/// ranking failures.
pub fn search_workspace(
database_path: &Path,
workspace: &str,
input: &SearchInput,
) -> Result<ToolEnvelope<SearchReport>, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
let snapshot = store.current_snapshot_summary(workspace)?;
let input_json = serde_json::to_vec(input)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
let input_fingerprint = stable_id_bytes("query-cache", &input_json);
let now_unix_ms = current_unix_millis();
if let Some(cached) = store.load_query_cache(
workspace,
&snapshot.snapshot_id,
&input_fingerprint,
now_unix_ms,
)? && let Ok(envelope) =
serde_json::from_slice::<ToolEnvelope<SearchReport>>(&cached.result_summary_json)
{
return Ok(envelope);
}
let (nodes, edges) = store.load_current_graph(workspace)?;
let repository_freshness = store.load_current_freshness(workspace)?;
let freshness = freshness_summary(&repository_freshness);
let fts_hits = if input.query.trim().is_empty() {
Vec::new()
} else {
store.search_current_nodes_ranked(workspace, &input.query, 500)?
};
let community_snapshot = store.load_current_community_snapshot(workspace)?;
let evidence = store.load_current_evidence(workspace)?;
let request = SearchRequest {
query: input.query.clone(),
filters: SearchFilters {
node_kinds: input.node_kinds.clone(),
repo_ids: input.repo_ids.clone(),
workspace_nodes: Vec::new(),
service_ids: input.service_ids.clone(),
community_ids: input.community_ids.clone(),
},
fts_scores: fts_search_scores(&fts_hits),
centrality_scores: community_centrality_scores(&community_snapshot.communities),
service_memberships: service_memberships(&nodes, &edges),
community_memberships: community_memberships(&community_snapshot.communities),
evidence: node_evidence(&edges, &evidence),
freshness: repository_freshness
.iter()
.map(|item| (item.repo_id.clone(), item.state))
.collect(),
offset: input.offset,
limit: input.limit,
};
let report = search(&nodes, &request)?;
let status = if report.coverage.gaps.is_empty() && freshness.overall == OverallFreshness::Fresh
{
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
let envelope = ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness,
warnings: Vec::new(),
};
let encoded = serde_json::to_vec(&envelope)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
let envelope = serde_json::from_slice(&encoded)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
drop(store);
if let Ok(_lock) = StoreLock::acquire(database_path, Duration::from_secs(5))
&& let Ok(mut writable) = SqliteStore::open(database_path)
{
let _ = writable.put_query_cache(&QueryCacheRecord {
workspace_name: workspace.to_owned(),
snapshot_id: snapshot.snapshot_id,
input_fingerprint,
result_summary_json: encoded,
stored_at_unix_ms: now_unix_ms,
expires_at_unix_ms: None,
});
}
Ok(envelope)
}
/// Executes an advanced bounded traversal against the current immutable graph.
///
/// # Errors
///
/// Returns [`ApplicationError`] for missing snapshots, invalid anchors, or bounds beyond the
/// server-enforced maximums.
pub fn traverse_workspace(
database_path: &Path,
workspace: &str,
input: &TraversalRequest,
) -> Result<ToolEnvelope<TraversalReport>, ApplicationError> {
if input.options.max_depth > 32
|| input.options.max_cross_repo_hops > 16
|| input.options.node_limit > 50_000
|| input.options.edge_limit > 250_000
|| input.options.timeout_ms > 5_000
|| input.options.k > 8
{
return Err(QueryError::InvalidTraversalLimits.into());
}
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let freshness = freshness_summary(&store.load_current_freshness(workspace)?);
let report = traverse(&nodes, &edges, input)?;
let status = if report.coverage_gaps.is_empty()
&& !report.truncated
&& freshness.overall == OverallFreshness::Fresh
{
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness,
warnings: Vec::new(),
})
}
/// Lists, inspects, or compares persisted communities for the current workspace.
///
/// # Errors
///
/// Returns [`ApplicationError`] for invalid pagination, missing snapshots, or malformed storage.
pub fn communities_workspace(
database_path: &Path,
workspace: &str,
input: &CommunityInput,
) -> Result<ToolEnvelope<CommunityReport>, ApplicationError> {
if !(1..=100).contains(&input.limit) || input.offset > 1_000_000 {
return Err(QueryError::InvalidPagination.into());
}
let store = SqliteStore::open_read_only(database_path)?;
let current = store.load_current_community_snapshot(workspace)?;
let freshness = freshness_summary(&store.load_current_freshness(workspace)?);
let mut selected = current
.communities
.iter()
.filter(|community| {
input
.community_id
.as_ref()
.is_none_or(|id| community.id == *id)
})
.cloned()
.collect::<Vec<_>>();
selected.sort_by(|left, right| {
right
.metrics
.size
.cmp(&left.metrics.size)
.then_with(|| right.metrics.cohesion.total_cmp(&left.metrics.cohesion))
.then_with(|| left.id.cmp(&right.id))
});
let total_communities = selected.len();
let communities = selected
.into_iter()
.skip(input.offset)
.take(input.limit)
.collect();
let delta = input
.compare_snapshot_id
.as_ref()
.map(|snapshot_id| store.load_workspace_community_snapshot(workspace, snapshot_id))
.transpose()?
.map(|before| compare_community_snapshots(&before, ¤t));
let exact_missing = input.community_id.is_some() && total_communities == 0;
let truncated = input.offset.saturating_add(input.limit) < total_communities;
let report = CommunityReport {
snapshot_id: current.snapshot_id,
engine_version: current.engine_version,
config: current.config,
communities,
total_communities,
offset: input.offset,
limit: input.limit,
truncated,
delta,
};
let status = if !exact_missing && freshness.overall == OverallFreshness::Fresh {
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness,
warnings: exact_missing
.then(|| "Requested community was not observed in the selected snapshot.".to_owned())
.into_iter()
.collect(),
})
}
/// Explores one registered checkout through bounded, ephemeral local code intelligence.
///
/// The returned [`LocalContextResult::content`] may contain source code and must never be
/// persisted, logged, cached, or included in audit records.
pub async fn explore_repository(
database_path: &Path,
workspace: &str,
input: &ExploreInput,
binary: Option<std::ffi::OsString>,
) -> ToolEnvelope<LocalContextResult> {
if input.workspace != workspace {
return explore_error_envelope(format!(
"workspace `{}` is outside this server's configured workspace `{workspace}`",
input.workspace
));
}
if !(1..=25).contains(&input.max_files) {
return explore_error_envelope("max_files must be between 1 and 25".to_owned());
}
let store = match SqliteStore::open_read_only(database_path) {
Ok(store) => store,
Err(error) => return explore_error_envelope(error.to_string()),
};
let registry = match store.load_workspace_registry(workspace) {
Ok(registry) => registry,
Err(error) => return explore_error_envelope(error.to_string()),
};
let persisted_freshness = match store.load_current_freshness(workspace) {
Ok(freshness) => freshness,
Err(error) => return explore_error_envelope(error.to_string()),
};
let freshness = freshness_summary(&persisted_freshness);
let requested_alias = input
.repository
.as_deref()
.map(str::trim)
.filter(|alias| !alias.is_empty());
let repository = match select_explore_repository(®istry, requested_alias, workspace) {
Ok(repository) => repository,
Err(message) => {
return explore_scoped_error_envelope(freshness, ToolStatus::Error, message);
}
};
let mut config = CodeGraphConfig::default();
if let Some(binary) = binary {
config.binary = binary;
}
let provider = match CodeGraphProvider::new(config) {
Ok(provider) => provider,
Err(error) => {
return explore_scoped_error_envelope(
freshness,
ToolStatus::Degraded,
error.to_string(),
);
}
};
let result = provider
.build_local_context(LocalContextRequest {
request: ProviderRequest {
repo_id: repository.id.clone(),
project_path: native_relative_path(&repository.canonical_path),
budget: ProviderBudget {
timeout: Duration::from_secs(5),
max_output_bytes: 256 * 1024,
max_items: 25,
},
cancellation: CancellationToken::new(),
},
query: input.query.clone(),
max_files: input.max_files,
})
.await;
let shutdown = provider.shutdown().await;
explore_result_envelope(result, shutdown, freshness)
}
fn select_explore_repository<'a>(
registry: &'a WorkspaceRecord,
requested_alias: Option<&str>,
workspace: &str,
) -> Result<&'a RepositoryRecord, String> {
if let Some(alias) = requested_alias {
return registry
.repositories
.iter()
.find(|repository| repository.alias == alias)
.ok_or_else(|| {
format!("repository alias `{alias}` is not registered in workspace `{workspace}`")
});
}
if registry.repositories.len() == 1 {
return registry.repositories.first().ok_or_else(|| {
format!("workspace `{workspace}` unexpectedly has no registered repository")
});
}
Err(format!(
"repository is required because workspace `{workspace}` contains {} repositories",
registry.repositories.len()
))
}
fn explore_result_envelope(
result: Result<LocalContextResult, ProviderError>,
shutdown: Result<(), ProviderError>,
freshness: FreshnessSummary,
) -> ToolEnvelope<LocalContextResult> {
match result {
Ok(context) => {
let mut warnings = context
.execution
.degradations
.iter()
.map(|degradation| degradation.message.clone())
.collect::<Vec<_>>();
if context.execution.truncated {
warnings.push("Local context was truncated by the configured budget.".to_owned());
}
if let Err(error) = shutdown {
warnings.push(format!("CodeGraph shutdown degraded: {error}"));
}
let status = if warnings.is_empty() && freshness.overall == OverallFreshness::Fresh {
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
ToolEnvelope {
schema_version: 1,
status,
data: Some(context),
freshness,
warnings,
}
}
Err(error) => ToolEnvelope {
schema_version: 1,
status: if matches!(&error, ProviderError::InvalidRequest(_)) {
ToolStatus::Error
} else {
ToolStatus::Degraded
},
data: None,
freshness,
warnings: vec![error.to_string()],
},
}
}
fn explore_scoped_error_envelope(
freshness: FreshnessSummary,
status: ToolStatus,
message: String,
) -> ToolEnvelope<LocalContextResult> {
ToolEnvelope {
schema_version: 1,
status,
data: None,
freshness,
warnings: vec![message],
}
}
fn explore_error_envelope(message: String) -> ToolEnvelope<LocalContextResult> {
ToolEnvelope {
schema_version: 1,
status: ToolStatus::Error,
data: None,
freshness: FreshnessSummary {
overall: OverallFreshness::Unknown,
stale_repositories: Vec::new(),
reasons: vec!["Local repository exploration could not start.".to_owned()],
},
warnings: vec![message],
}
}
/// Computes conservative upstream or downstream impact from the current immutable graph.
///
/// The default delivery path uses persisted graph, freshness, community, test, and ownership data.
/// Optional compatibility and local-provider enrichment are added by specialized callers before
/// invoking the pure core engine.
///
/// # Errors
///
/// Returns [`ApplicationError`] for missing snapshots, malformed persisted data, unknown targets,
/// dangling graph relationships, or invalid analysis bounds.
pub fn impact_workspace(
database_path: &Path,
workspace: &str,
request: &ImpactRequest,
) -> Result<ToolEnvelope<ImpactReport>, ApplicationError> {
let loaded = load_impact_context(database_path, workspace)?;
impact_envelope(request, &loaded.context, loaded.freshness)
}
/// Computes impact with optional bounded, ephemeral `CodeGraph` local enrichment.
///
/// Provider failures, stale indexes, missing capabilities, and truncation are converted into
/// visible coverage degradation and can never lower the federated risk.
///
/// # Errors
///
/// Returns [`ApplicationError`] for persisted graph failures or invalid impact inputs. `CodeGraph`
/// operational failures degrade the report instead of failing the federated analysis.
pub async fn impact_workspace_with_codegraph(
database_path: &Path,
workspace: &str,
request: &ImpactRequest,
binary: Option<std::ffi::OsString>,
) -> Result<ToolEnvelope<ImpactReport>, ApplicationError> {
let mut loaded = load_impact_context(database_path, workspace)?;
let mut config = CodeGraphConfig::default();
if let Some(binary) = binary {
config.binary = binary;
}
match CodeGraphProvider::new(config) {
Ok(provider) => {
enrich_impact_context(
&mut loaded.context,
request,
&loaded.registry,
&loaded.node_evidence,
&provider,
)
.await;
let _ = provider.shutdown().await;
}
Err(error) => loaded.context.local_enrichment.push(LocalEnrichmentInput {
repo_id: impact_target_repo(&loaded.context, request)
.unwrap_or_else(|| RepoId::new("repo:unknown")),
anchor: impact_target_label(&loaded.context, request)
.unwrap_or_else(|| "unknown impact target".to_owned()),
status: LocalEnrichmentStatus::Unavailable,
affected: Vec::new(),
affected_tests: Vec::new(),
truncated: false,
degradations: vec![error.to_string()],
}),
}
impact_envelope(request, &loaded.context, loaded.freshness)
}
/// Collects a bounded fingerprinted local Git change set for one registered checkout.
///
/// This operation never stages, commits, modifies, or initializes repository state.
///
/// # Errors
///
/// Returns [`ApplicationError`] for unknown repositories, missing snapshots, invalid scopes,
/// unsafe refs/pathspecs, Git failures, cancellation, timeouts, or output-budget exhaustion.
pub async fn collect_workspace_changes(
database_path: &Path,
workspace: &str,
input: &ChangesInput,
git_binary: Option<std::ffi::OsString>,
) -> Result<ToolEnvelope<ChangeSet>, ApplicationError> {
collect_workspace_changes_with_cancellation(
database_path,
workspace,
input,
git_binary,
&CancellationToken::new(),
)
.await
}
/// Collects a local change set with caller-owned cooperative cancellation.
///
/// # Errors
///
/// Returns the same failures as [`collect_workspace_changes`], including cancellation.
pub async fn collect_workspace_changes_with_cancellation(
database_path: &Path,
workspace: &str,
input: &ChangesInput,
git_binary: Option<std::ffi::OsString>,
cancellation: &CancellationToken,
) -> Result<ToolEnvelope<ChangeSet>, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
let registry = store.load_workspace_registry(workspace)?;
let snapshot = store.current_snapshot_summary(workspace)?;
let repository = registry
.repositories
.iter()
.find(|repository| repository.alias == input.repository)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(input.repository.clone()))?;
let persisted_freshness = store.load_current_freshness(workspace)?;
let freshness = freshness_summary(&persisted_freshness);
let provider = git_binary.map_or_else(GitCliChangeProvider::new, |binary| {
GitCliChangeProvider::with_limits(
binary,
Duration::from_secs(30),
8 * 1024 * 1024,
64 * 1024,
)
});
let mut analyzer_versions = AnalyzerVersions::new();
analyzer_versions.insert(
"code-system-graph.changes".to_owned(),
env!("CARGO_PKG_VERSION").to_owned(),
);
let change_set = provider
.changes(
&ChangeRequest {
repo_id: repository.id.clone(),
checkout_id: repository.checkout_id.clone(),
worktree: native_relative_path(&repository.canonical_path),
scope: input.scope.clone(),
workspace_manifest_hash: registry.manifest_hash.clone(),
contract_registry_hash: snapshot.snapshot_id,
analyzer_versions,
},
cancellation,
)
.await?;
let repository_is_fresh = persisted_freshness.iter().any(|item| {
item.checkout_id == repository.checkout_id && item.state == RepoFreshnessState::Fresh
});
Ok(ToolEnvelope {
schema_version: 1,
status: if repository_is_fresh {
ToolStatus::Ok
} else {
ToolStatus::Degraded
},
data: Some(change_set),
freshness,
warnings: Vec::new(),
})
}
/// Collects local Git state and maps it to bounded semantic graph impact.
///
/// # Errors
///
/// Returns [`ApplicationError`] for change collection, storage, graph validation, or semantic
/// analysis failures.
pub async fn analyze_workspace_changes(
database_path: &Path,
workspace: &str,
input: &ChangesInput,
options: &ChangeAnalysisOptions,
git_binary: Option<std::ffi::OsString>,
) -> Result<ToolEnvelope<ChangeImpactReport>, ApplicationError> {
analyze_workspace_changes_with_cancellation(
database_path,
workspace,
input,
options,
git_binary,
&CancellationToken::new(),
)
.await
}
/// Analyzes local changes with caller-owned cooperative cancellation.
///
/// # Errors
///
/// Returns the same failures as [`analyze_workspace_changes`], including cancellation.
pub async fn analyze_workspace_changes_with_cancellation(
database_path: &Path,
workspace: &str,
input: &ChangesInput,
options: &ChangeAnalysisOptions,
git_binary: Option<std::ffi::OsString>,
cancellation: &CancellationToken,
) -> Result<ToolEnvelope<ChangeImpactReport>, ApplicationError> {
let collected = collect_workspace_changes_with_cancellation(
database_path,
workspace,
input,
git_binary,
cancellation,
)
.await?;
let change_set = collected.data.ok_or_else(|| {
ApplicationError::InvalidSourceObservation(
"local change collection returned no change set".to_owned(),
)
})?;
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let evidence = store.load_current_evidence(workspace)?;
let community = store.load_current_community_snapshot(workspace)?;
let freshness = store.load_current_freshness(workspace)?;
let report = analyze_changes(
&change_set,
&nodes,
&edges,
&evidence,
Some(&community),
&freshness,
&[],
options,
)?;
let status = if collected.status == ToolStatus::Ok && report.coverage.complete {
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness: collected.freshness,
warnings: collected.warnings,
})
}
fn pull_request_provider(
kind: PullRequestProviderKind,
enabled: bool,
token: Option<String>,
basic_auth_username: Option<String>,
) -> Result<Box<dyn PullRequestProvider>, ApplicationError> {
if !enabled {
return Err(PullRequestError::Disabled.into());
}
let mut config = match kind {
PullRequestProviderKind::GitHub => PullRequestProviderConfig::default(),
PullRequestProviderKind::BitbucketCloud => PullRequestProviderConfig::bitbucket_cloud(),
PullRequestProviderKind::BitbucketDataCenter => {
return Err(PullRequestError::InvalidConfiguration(
"Bitbucket Data Center is not implemented in Code System Graph 1.0.0".to_owned(),
)
.into());
}
};
config.enabled = true;
config.auth_token = PrAuthToken::new(token.unwrap_or_default());
if kind == PullRequestProviderKind::BitbucketCloud {
config.basic_auth_username = basic_auth_username;
}
let transport = Arc::new(
ReqwestPrHttpTransport::new()
.map_err(|error| PullRequestError::Transport(error.to_string()))?,
);
match kind {
PullRequestProviderKind::GitHub => Ok(Box::new(GitHubProvider::new(config, transport)?)),
PullRequestProviderKind::BitbucketCloud => {
Ok(Box::new(BitbucketProvider::new(config, transport)?))
}
PullRequestProviderKind::BitbucketDataCenter => unreachable!("rejected above"),
}
}
/// Inspects one GitHub or Bitbucket Cloud pull request through the public provider API.
///
/// Both the provider-level `enabled` gate and per-request consent must be true. The token remains
/// ephemeral and is excluded from all returned data.
///
/// # Errors
///
/// Returns [`ApplicationError`] for disabled providers, missing consent, unsafe configuration,
/// cancellation, timeouts, rate limits, provider errors, malformed responses, or unsupported
/// Bitbucket Data Center requests.
pub async fn inspect_pull_request(
input: &PullRequestInput,
enabled: bool,
token: Option<String>,
basic_auth_username: Option<String>,
) -> Result<ToolEnvelope<PullRequestInspection>, ApplicationError> {
inspect_pull_request_with_cancellation(
input,
enabled,
token,
basic_auth_username,
CancellationToken::new(),
)
.await
}
/// Inspects one pull request with caller-owned cooperative cancellation.
///
/// # Errors
///
/// Returns the same failures as [`inspect_pull_request`], including cancellation.
pub async fn inspect_pull_request_with_cancellation(
input: &PullRequestInput,
enabled: bool,
token: Option<String>,
basic_auth_username: Option<String>,
cancellation: CancellationToken,
) -> Result<ToolEnvelope<PullRequestInspection>, ApplicationError> {
if !input.consent_to_remote_access {
return Err(PullRequestError::ConsentRequired.into());
}
let provider = pull_request_provider(input.provider, enabled, token, basic_auth_username)?;
let inspection = provider
.inspect(PullRequestInspectRequest {
coordinates: PullRequestCoordinates {
provider: input.provider,
owner: input.owner.clone(),
repository: input.repository.clone(),
number: input.number,
},
consent_to_remote_access: input.consent_to_remote_access,
cancellation,
})
.await?;
let status = if inspection.warnings.is_empty() {
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(inspection),
freshness: FreshnessSummary {
overall: OverallFreshness::Unknown,
stale_repositories: Vec::new(),
reasons: vec![
"Remote pull-request inspection does not establish local graph freshness."
.to_owned(),
],
},
warnings: Vec::new(),
})
}
/// Lists one bounded page of source-free hosted pull-request summaries.
///
/// # Errors
///
/// Returns [`ApplicationError`] for disabled providers, missing consent, invalid bounds or
/// cursors, cancellation, timeouts, rate limits, malformed responses, or unsupported providers.
pub async fn list_pull_requests(
input: &PullRequestListInput,
enabled: bool,
token: Option<String>,
basic_auth_username: Option<String>,
cancellation: CancellationToken,
) -> Result<ToolEnvelope<PullRequestListPage>, ApplicationError> {
if !input.consent_to_remote_access {
return Err(PullRequestError::ConsentRequired.into());
}
let provider = pull_request_provider(input.provider, enabled, token, basic_auth_username)?;
let page = provider
.list(PullRequestListRequest {
provider: input.provider,
owner: input.owner.clone(),
repository: input.repository.clone(),
state: input.state,
cursor: input.cursor.clone(),
limit: input.limit,
consent_to_remote_access: input.consent_to_remote_access,
cancellation,
})
.await?;
let status = if page.warnings.is_empty() && !page.truncated {
ToolStatus::Ok
} else {
ToolStatus::Degraded
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(page),
freshness: FreshnessSummary {
overall: OverallFreshness::Unknown,
stale_repositories: Vec::new(),
reasons: vec![
"Remote pull-request listing does not establish local graph freshness.".to_owned(),
],
},
warnings: Vec::new(),
})
}
struct LoadedImpactContext {
context: ImpactContext,
freshness: FreshnessSummary,
registry: WorkspaceRecord,
node_evidence: BTreeMap<NodeId, Vec<Evidence>>,
}
fn load_impact_context(
database_path: &Path,
workspace: &str,
) -> Result<LoadedImpactContext, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let persisted_freshness = store.load_current_freshness(workspace)?;
let freshness = freshness_summary(&persisted_freshness);
let communities = store.load_current_community_snapshot(workspace)?;
let evidence = store.load_current_evidence(workspace)?;
let registry = store.load_workspace_registry(workspace)?;
let node_evidence = node_evidence(&edges, &evidence);
let context = ImpactContext {
centrality: impact_centrality_scores(&communities.communities),
service_memberships: service_memberships(&nodes, &edges),
recommended_commands: recommended_test_commands(&nodes, &node_evidence),
nodes,
edges,
communities: Some(communities),
freshness: conservative_repo_freshness(&persisted_freshness),
compatibility: Vec::new(),
local_enrichment: Vec::new(),
public_contracts: Vec::new(),
criticality: Vec::new(),
environments: Vec::new(),
graph_complete: true,
coverage_gaps: Vec::new(),
};
Ok(LoadedImpactContext {
context,
freshness,
registry,
node_evidence,
})
}
fn impact_envelope(
request: &ImpactRequest,
context: &ImpactContext,
freshness: FreshnessSummary,
) -> Result<ToolEnvelope<ImpactReport>, ApplicationError> {
let report = analyze_impact(request, context)?;
let status = if report.risk == code_system_graph_core::RiskLevel::Unknown
|| freshness.overall != OverallFreshness::Fresh
{
ToolStatus::Degraded
} else {
ToolStatus::Ok
};
Ok(ToolEnvelope {
schema_version: 1,
status,
data: Some(report),
freshness,
warnings: Vec::new(),
})
}
#[expect(
clippy::too_many_lines,
reason = "Provider probing, impact, affected tests, and degradation form one bounded lifecycle"
)]
async fn enrich_impact_context(
context: &mut ImpactContext,
request: &ImpactRequest,
registry: &WorkspaceRecord,
evidence: &BTreeMap<NodeId, Vec<Evidence>>,
provider: &impl LocalCodeIntelligenceProvider,
) {
let Some(target) = impact_target_node(context, request).cloned() else {
return;
};
let by_id = context
.nodes
.iter()
.map(|node| (node.id.clone(), node))
.collect::<BTreeMap<_, _>>();
let mut anchors = context
.edges
.iter()
.filter_map(|edge| {
let adjacent = if edge.source == target.id {
by_id.get(&edge.target)
} else if edge.target == target.id {
by_id.get(&edge.source)
} else {
None
};
adjacent
.copied()
.filter(|node| node.kind == NodeKind::SymbolRef)
.cloned()
})
.chain((target.kind == NodeKind::SymbolRef).then_some(target.clone()))
.collect::<Vec<_>>();
anchors.sort_by(|left, right| left.id.cmp(&right.id));
anchors.dedup_by(|left, right| left.id == right.id);
anchors.truncate(3);
for anchor in anchors {
let Some(repo_id) = anchor.repo_id.clone() else {
continue;
};
let Some(repository) = registry
.repositories
.iter()
.find(|repository| repository.id == repo_id)
else {
context.local_enrichment.push(unavailable_local_enrichment(
repo_id,
&anchor.label,
"Registered checkout metadata was unavailable for local enrichment.",
));
continue;
};
let project_path = native_relative_path(&repository.canonical_path);
let provider_request = || ProviderRequest {
repo_id: repo_id.clone(),
project_path: project_path.clone(),
budget: ProviderBudget {
timeout: Duration::from_secs(2),
max_output_bytes: 256 * 1024,
max_items: 25,
},
cancellation: CancellationToken::new(),
};
let capability = match provider.probe(provider_request()).await {
Ok(capability) => capability,
Err(error) => {
context.local_enrichment.push(unavailable_local_enrichment(
repo_id,
&anchor.label,
&error.to_string(),
));
continue;
}
};
if capability.status != ProviderStatus::Available {
let status = if capability.status == ProviderStatus::Stale {
LocalEnrichmentStatus::Stale
} else {
LocalEnrichmentStatus::Unavailable
};
let mut degradations = capability
.degradations
.into_iter()
.map(|degradation| degradation.message)
.collect::<Vec<_>>();
degradations.extend(capability.remediation);
context.local_enrichment.push(LocalEnrichmentInput {
repo_id,
anchor: anchor.label.clone(),
status,
affected: Vec::new(),
affected_tests: Vec::new(),
truncated: false,
degradations,
});
continue;
}
let result = match provider
.get_local_impact(LocalImpactRequest {
request: provider_request(),
symbol: anchor.label.clone(),
max_depth: request.options.max_depth.min(8),
})
.await
{
Ok(result) => result,
Err(error) => {
context.local_enrichment.push(unavailable_local_enrichment(
repo_id,
&anchor.label,
&error.to_string(),
));
continue;
}
};
let changed_files = evidence
.get(&anchor.id)
.into_iter()
.flatten()
.filter_map(|item| item.file_path.clone())
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let affected_tests = if changed_files.is_empty() {
None
} else {
provider
.get_affected_tests(AffectedTestsRequest {
request: provider_request(),
changed_files,
max_depth: request.options.max_depth.min(8),
})
.await
.ok()
.flatten()
};
let mut degradations = result
.execution
.degradations
.iter()
.map(|degradation| degradation.message.clone())
.collect::<Vec<_>>();
if affected_tests.is_none() {
degradations.push(
"CodeGraph affected-test capability was unavailable or no changed file was known."
.to_owned(),
);
}
let tests_truncated = affected_tests
.as_ref()
.is_some_and(|tests| tests.execution.truncated);
let status = if result.execution.truncated || tests_truncated || !degradations.is_empty() {
LocalEnrichmentStatus::Partial
} else {
LocalEnrichmentStatus::Available
};
context.local_enrichment.push(LocalEnrichmentInput {
repo_id,
anchor: result.symbol,
status,
affected: result
.affected
.into_iter()
.map(|item| LocalImpactItem {
symbol: item.name,
file_path: item.file_path,
start_line: Some(item.start_line),
depth: result.depth,
})
.collect(),
affected_tests: affected_tests
.map(|tests| tests.affected_tests)
.unwrap_or_default(),
truncated: result.execution.truncated || tests_truncated,
degradations,
});
}
}
fn unavailable_local_enrichment(
repo_id: RepoId,
anchor: &str,
message: &str,
) -> LocalEnrichmentInput {
LocalEnrichmentInput {
repo_id,
anchor: anchor.to_owned(),
status: LocalEnrichmentStatus::Unavailable,
affected: Vec::new(),
affected_tests: Vec::new(),
truncated: false,
degradations: vec![message.to_owned()],
}
}
fn impact_target_node<'a>(context: &'a ImpactContext, request: &ImpactRequest) -> Option<&'a Node> {
context.nodes.iter().find(|node| match &request.target {
ImpactTarget::NodeId(node_id) => node.id == *node_id,
ImpactTarget::StableKey(stable_key) => node.stable_key == *stable_key,
})
}
fn impact_target_repo(context: &ImpactContext, request: &ImpactRequest) -> Option<RepoId> {
impact_target_node(context, request).and_then(|node| node.repo_id.clone())
}
fn impact_target_label(context: &ImpactContext, request: &ImpactRequest) -> Option<String> {
impact_target_node(context, request).map(|node| node.label.clone())
}
fn fts_search_scores(
hits: &[code_system_graph_store_sqlite::StoredNodeSearchHit],
) -> BTreeMap<NodeId, f64> {
hits.iter()
.enumerate()
.map(|(index, hit)| {
(
hit.node.id.clone(),
1.0 / f64::from(u32::try_from(index + 1).unwrap_or(u32::MAX)),
)
})
.collect()
}
fn impact_centrality_scores(communities: &[Community]) -> BTreeMap<NodeId, f32> {
let mut scores = BTreeMap::<NodeId, f32>::new();
for community in communities {
for (index, node_id) in community.central_nodes.iter().enumerate() {
let denominator = u16::try_from(index + 1).unwrap_or(u16::MAX);
let score = 1.0 / f32::from(denominator);
scores
.entry(node_id.clone())
.and_modify(|current| *current = current.max(score))
.or_insert(score);
}
}
scores
}
fn conservative_repo_freshness(items: &[RepoFreshness]) -> Vec<RepoFreshness> {
let mut by_repository = BTreeMap::<RepoId, RepoFreshness>::new();
for item in items {
by_repository
.entry(item.repo_id.clone())
.and_modify(|current| {
if freshness_severity(item.state) > freshness_severity(current.state) {
*current = item.clone();
}
})
.or_insert_with(|| item.clone());
}
by_repository.into_values().collect()
}
fn freshness_severity(state: RepoFreshnessState) -> u8 {
match state {
RepoFreshnessState::Fresh => 0,
RepoFreshnessState::WorkingTreeChanged
| RepoFreshnessState::CommitsBehind
| RepoFreshnessState::ConfigChanged
| RepoFreshnessState::ExtractorChanged
| RepoFreshnessState::CodegraphPending => 1,
RepoFreshnessState::Partial | RepoFreshnessState::Unknown => 2,
RepoFreshnessState::Unavailable => 3,
RepoFreshnessState::Corrupt => 4,
}
}
fn recommended_test_commands(
nodes: &[Node],
evidence: &BTreeMap<NodeId, Vec<Evidence>>,
) -> Vec<RecommendedCommand> {
let mut commands = BTreeMap::<(RepoId, String), RecommendedCommand>::new();
for test in nodes.iter().filter(|node| node.kind == NodeKind::TestCase) {
let Some(repo_id) = test.repo_id.clone() else {
continue;
};
for item in evidence.get(&test.id).into_iter().flatten() {
let Some(path) = item.file_path.as_deref() else {
continue;
};
let command = match Path::new(path).extension().and_then(|value| value.to_str()) {
Some(extension) if extension.eq_ignore_ascii_case("py") => "python -m pytest",
Some(extension) if extension.eq_ignore_ascii_case("rs") => "cargo test",
_ => continue,
};
commands.entry((repo_id.clone(), command.to_owned())).or_insert_with(|| RecommendedCommand {
repo_id: repo_id.clone(),
command: command.to_owned(),
description: format!("Run the repository test suite covering `{}`; Code System Graph does not execute it.", test.label),
});
}
}
commands.into_values().collect()
}
fn community_centrality_scores(communities: &[Community]) -> BTreeMap<NodeId, f64> {
let mut scores = BTreeMap::<NodeId, f64>::new();
for community in communities {
for (index, node_id) in community.central_nodes.iter().enumerate() {
let score = 1.0 / f64::from(u32::try_from(index + 1).unwrap_or(u32::MAX));
scores
.entry(node_id.clone())
.and_modify(|current| *current = current.max(score))
.or_insert(score);
}
}
scores
}
fn community_memberships(communities: &[Community]) -> BTreeMap<NodeId, Vec<CommunityId>> {
let mut memberships = BTreeMap::<NodeId, Vec<CommunityId>>::new();
for community in communities {
for member in &community.members {
memberships
.entry(member.clone())
.or_default()
.push(community.id.clone());
}
}
memberships
}
fn service_memberships(nodes: &[Node], edges: &[Edge]) -> BTreeMap<NodeId, Vec<NodeId>> {
let services = nodes
.iter()
.filter(|node| node.kind == NodeKind::Service)
.map(|node| node.id.clone())
.collect::<BTreeSet<_>>();
let mut memberships = services
.iter()
.map(|service| (service.clone(), vec![service.clone()]))
.collect::<BTreeMap<_, _>>();
for edge in edges {
if services.contains(&edge.source) {
memberships
.entry(edge.target.clone())
.or_default()
.push(edge.source.clone());
}
if services.contains(&edge.target) {
memberships
.entry(edge.source.clone())
.or_default()
.push(edge.target.clone());
}
}
for values in memberships.values_mut() {
values.sort();
values.dedup();
}
memberships
}
fn node_evidence(edges: &[Edge], evidence: &[Evidence]) -> BTreeMap<NodeId, Vec<Evidence>> {
let by_id = evidence
.iter()
.map(|item| (item.id.clone(), item))
.collect::<BTreeMap<_, _>>();
let mut result = BTreeMap::<NodeId, Vec<Evidence>>::new();
for edge in edges {
for node_id in [&edge.source, &edge.target] {
let values = result.entry(node_id.clone()).or_default();
values.extend(
edge.evidence
.iter()
.filter_map(|evidence_id| by_id.get(evidence_id))
.map(|item| (*item).clone()),
);
}
}
for values in result.values_mut() {
values.sort_by(|left, right| left.id.cmp(&right.id));
values.dedup_by(|left, right| left.id == right.id);
}
result
}
/// Evaluates current repository state against the published snapshot.
///
/// # Errors
///
/// Returns [`ApplicationError`] for invalid manifests, path-policy failures, missing snapshots,
/// or storage failures.
pub fn status_workspace(
config_path: &Path,
database_path: &Path,
) -> Result<WorkspaceStatus, ApplicationError> {
let current = load_workspace_context(config_path, &ScanOverrides::default())?;
let store = SqliteStore::open_read_only(database_path)?;
let previous = store.load_current_freshness(¤t.manifest.name)?;
let previous_by_checkout = previous
.iter()
.map(|freshness| (freshness.checkout_id.clone(), freshness))
.collect::<BTreeMap<CheckoutId, _>>();
let current_checkouts = current
.registry
.record
.repositories
.iter()
.map(|repository| repository.checkout_id.clone())
.collect::<std::collections::BTreeSet<_>>();
let mut repositories = current
.registry
.record
.repositories
.iter()
.map(|repository| {
let previous = previous_by_checkout.get(&repository.checkout_id);
let (state, reason) = evaluate_repository_freshness(
repository,
previous.copied(),
¤t.registry.record.manifest_hash,
);
RepoFreshness {
repo_id: repository.id.clone(),
checkout_id: repository.checkout_id.clone(),
head_commit: repository.head_commit.clone(),
manifest_hash: current.registry.record.manifest_hash.clone(),
state,
reason,
}
})
.collect::<Vec<_>>();
repositories.extend(
previous
.into_iter()
.filter(|freshness| !current_checkouts.contains(&freshness.checkout_id))
.map(|mut freshness| {
freshness.state = RepoFreshnessState::Unavailable;
freshness.reason = Some(
"Checkout from the published snapshot is absent from the current manifest."
.to_owned(),
);
freshness
}),
);
repositories.sort_by(|left, right| {
(&left.repo_id, &left.checkout_id).cmp(&(&right.repo_id, &right.checkout_id))
});
let watcher = watcher_status(
database_path,
¤t.manifest.name,
current.execution_policy.max_no_progress_time_ms,
&store.database_instance_id()?,
);
Ok(WorkspaceStatus {
workspace: current.manifest.name,
schema_version: store.schema_version()?,
integrity_ok: store.integrity_check()?,
freshness: freshness_summary(&repositories),
repositories,
watcher,
})
}
fn watcher_status(
database: &Path,
workspace: &str,
stale_after_ms: u64,
database_instance_id: &str,
) -> WatcherStatus {
if !work_state::work_path(database).exists() {
return WatcherStatus {
state: WatcherState::NeverStarted,
detail: None,
};
}
let lease = work_state::WorkState::open(database, database_instance_id)
.and_then(|state| state.watcher_lease(workspace));
let Ok(Some(lease)) = lease else {
return WatcherStatus {
state: WatcherState::NeverStarted,
detail: None,
};
};
let persisted = match lease.state.as_str() {
"expired_idle" => Some(WatcherState::ExpiredIdle),
"expired_session" => Some(WatcherState::ExpiredSession),
"failed_limit" => Some(WatcherState::FailedLimit),
"active" => None,
_ => Some(WatcherState::Stale),
};
if let Some(state) = persisted {
return WatcherStatus {
state,
detail: lease.detail,
};
}
let identity_matches = lease
.pid
.zip(lease.process_start_identity.as_deref())
.is_some_and(|(pid, expected)| worker::process_identity(pid).as_deref() == Some(expected));
let heartbeat_fresh = lease.heartbeat_unix_ms.is_some_and(|heartbeat| {
current_unix_millis().saturating_sub(heartbeat) <= stale_after_ms.saturating_mul(2)
});
if identity_matches && heartbeat_fresh {
WatcherStatus {
state: WatcherState::Active,
detail: lease.detail,
}
} else {
WatcherStatus {
state: WatcherState::Stale,
detail: Some("watcher heartbeat or process start identity is stale".to_owned()),
}
}
}
/// Initializes the operational lease used by the hidden finite-watcher implementation.
#[doc(hidden)]
pub fn start_watcher_lease(
config: &Path,
database: &Path,
) -> Result<(String, String, ExecutionPolicy), ApplicationError> {
let manifest_source = read_file(config)?;
let manifest = parse_manifest(&manifest_source)?;
validate_global_policy_source(config, &manifest)?;
let execution_policy = ExecutionPolicy::resolve(manifest.execution_policy.as_ref())
.map_err(ManifestError::from)?;
let identity = worker::process_identity(std::process::id()).ok_or_else(|| {
ApplicationError::Initialization("failed to resolve watcher process identity".to_owned())
})?;
let now = current_unix_millis();
let owner_token = stable_id(
"watcher-lease",
&format!("{}:{}:{identity}:{now}", manifest.name, std::process::id()),
);
let database_instance_id = SqliteStore::open(database)?.database_instance_id()?;
let result =
work_state::WorkState::open(database, &database_instance_id).and_then(|mut state| {
state.start_watcher(
&manifest.name,
&owner_token,
std::process::id(),
&identity,
now,
execution_policy.max_no_progress_time_ms,
)
});
if let Err(message) = result {
if message
== format!(
"watcher for workspace `{}` is already active",
manifest.name
)
{
return Err(ApplicationError::WatcherAlreadyActive(manifest.name));
}
return Err(ApplicationError::Initialization(message));
}
Ok((manifest.name, owner_token, execution_policy))
}
/// Renews the hidden finite-watcher heartbeat and, for relevant success, its idle lease.
#[doc(hidden)]
pub fn heartbeat_watcher_lease(
database: &Path,
workspace: &str,
owner_token: &str,
successful_activity: bool,
) -> Result<(), ApplicationError> {
let database_instance_id = work_database_instance_id(database)?;
work_state::WorkState::open(database, &database_instance_id)
.and_then(|state| {
state.heartbeat_watcher(
workspace,
owner_token,
current_unix_millis(),
successful_activity,
)
})
.map_err(ApplicationError::Initialization)
}
/// Persists one terminal hidden finite-watcher state.
#[doc(hidden)]
pub fn finish_watcher_lease(
database: &Path,
workspace: &str,
owner_token: &str,
state: &str,
detail: Option<&str>,
) -> Result<(), ApplicationError> {
let database_instance_id = work_database_instance_id(database)?;
work_state::WorkState::open(database, &database_instance_id)
.and_then(|work| {
work.finish_watcher(workspace, owner_token, state, detail, current_unix_millis())
})
.map_err(ApplicationError::Initialization)
}
fn work_database_instance_id(database: &Path) -> Result<String, ApplicationError> {
Ok(SqliteStore::open_read_only(database)?.database_instance_id()?)
}
/// Inspects, validates, compares, or explains contracts in the current immutable graph.
///
/// # Errors
///
/// Returns [`ApplicationError`] for missing snapshots, invalid stored data, or invalid bounded
/// contract requests.
pub fn contracts_workspace(
database_path: &Path,
workspace: &str,
request: &ContractRequest,
) -> Result<ContractReport, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let evidence = store.load_current_evidence(workspace)?;
Ok(inspect_contracts(&nodes, &edges, &evidence, &[], request)?)
}
/// Renders one deterministic bounded source-free graph export.
///
/// # Errors
///
/// Returns [`ApplicationError`] for missing snapshots, invalid stored data, or invalid bounds.
pub fn export_workspace(
database_path: &Path,
workspace: &str,
request: &ExportRequest,
) -> Result<ExportReport, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
let (nodes, edges) = store.load_current_graph(workspace)?;
let evidence = store.load_current_evidence(workspace)?;
Ok(export_graph(&nodes, &edges, &evidence, request)?)
}
/// Consolidates configuration, schema, integrity, and freshness diagnostics.
///
/// # Errors
///
/// Returns [`ApplicationError`] when required workspace or store observations cannot be loaded.
#[expect(
clippy::too_many_lines,
reason = "Doctor keeps every explicit safety observation visible in one conservative report"
)]
pub fn doctor_workspace(
config_path: &Path,
database_path: &Path,
) -> Result<DoctorReport, ApplicationError> {
let status = status_workspace(config_path, database_path)?;
let store = SqliteStore::open_read_only(database_path)?;
let diagnostics = store.diagnostics()?;
let context = load_workspace_context(config_path, &ScanOverrides::default())?;
let expected_version = u32::try_from(latest_schema_version()).map_err(|error| {
ApplicationError::Initialization(format!("unsupported schema version: {error}"))
})?;
let actual_version = u32::try_from(status.schema_version).ok();
let freshness = status
.repositories
.iter()
.map(|repository| FreshnessDoctorInput {
repo_id: repository.repo_id.clone(),
state: repository.state,
detail: repository.reason.clone(),
})
.collect();
let repository_paths_valid = context
.registry
.record
.repositories
.iter()
.all(|repository| {
context
.registry
.checkout_path(&repository.alias)
.is_some_and(Path::is_dir)
});
let git_available = std::process::Command::new("git")
.arg("--version")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.is_ok_and(|status| status.success());
Ok(doctor(&DoctorRequest {
schema: vec![SchemaDoctorInput {
name: "sqlite".to_owned(),
expected_version,
actual_version,
metadata_consistent: Some(actual_version == Some(expected_version)),
}],
integrity: vec![
IntegrityDoctorInput {
name: "binary-version".to_owned(),
passed: Some(true),
detail: Some(env!("CARGO_PKG_VERSION").to_owned()),
},
IntegrityDoctorInput {
name: "sqlite".to_owned(),
passed: Some(status.integrity_ok),
detail: None,
},
IntegrityDoctorInput {
name: "sqlite-wal".to_owned(),
passed: Some(diagnostics.journal_mode.eq_ignore_ascii_case("wal")),
detail: Some(diagnostics.journal_mode),
},
IntegrityDoctorInput {
name: "sqlite-foreign-keys".to_owned(),
passed: Some(diagnostics.foreign_keys_enabled),
detail: None,
},
IntegrityDoctorInput {
name: "sqlite-fts5".to_owned(),
passed: Some(diagnostics.fts5_index_available),
detail: None,
},
IntegrityDoctorInput {
name: "repository-paths".to_owned(),
passed: Some(repository_paths_valid),
detail: None,
},
IntegrityDoctorInput {
name: "git".to_owned(),
passed: Some(git_available),
detail: None,
},
IntegrityDoctorInput {
name: "extractor-matrix".to_owned(),
passed: Some(true),
detail: Some(
"Built-in extractor inventory is compiled into this binary.".to_owned(),
),
},
IntegrityDoctorInput {
name: "network-policy".to_owned(),
passed: Some(true),
detail: Some(
"Remote providers and non-loopback HTTP remain explicitly gated.".to_owned(),
),
},
IntegrityDoctorInput {
name: "writer-lock".to_owned(),
passed: None,
detail: Some(
"Read-only doctor does not acquire or reclaim the workspace writer lock."
.to_owned(),
),
},
IntegrityDoctorInput {
name: "hook-installation".to_owned(),
passed: None,
detail: Some(
"Use `csgraph hooks status` with an explicit host and repository root."
.to_owned(),
),
},
IntegrityDoctorInput {
name: "release-files".to_owned(),
passed: None,
detail: Some(
"Release artifact notices and checksums require an explicit installation path."
.to_owned(),
),
},
],
freshness,
providers: vec![ProviderDoctorInput {
name: "codegraph".to_owned(),
status: ProviderDoctorStatus::Unavailable,
detail: Some(
"CodeGraph probing is opt-in and was not requested by this doctor invocation."
.to_owned(),
),
}],
config: vec![ConfigDoctorInput {
name: status.workspace,
valid: Some(true),
detail: None,
}],
}))
}
/// Creates a redacted, source-free diagnostic bundle only at an explicit output path.
///
/// Existing files are never overwritten. On Unix, the file is created with owner-only
/// permissions before any diagnostic data is serialized.
///
/// # Errors
///
/// Returns [`ApplicationError`] when diagnostics cannot be collected, the destination already
/// exists, time cannot be represented, or the bundle cannot be durably written.
pub fn create_diagnostic_bundle(
config_path: &Path,
database_path: &Path,
output_path: &Path,
) -> Result<DiagnosticBundle, ApplicationError> {
let generated_at_unix_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?
.as_millis();
let bundle = DiagnosticBundle {
schema_version: 1,
binary_version: env!("CARGO_PKG_VERSION").to_owned(),
operating_system: std::env::consts::OS.to_owned(),
architecture: std::env::consts::ARCH.to_owned(),
generated_at_unix_ms,
debug_requested: std::env::var("CODE_SYSTEM_GRAPH_DEBUG")
.is_ok_and(|value| value.trim() == "1"),
doctor: doctor_workspace(config_path, database_path)?,
};
let mut options = OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
let mut output = options
.open(output_path)
.map_err(|source| ApplicationError::WriteFile {
path: output_path.to_path_buf(),
source,
})?;
serde_json::to_writer_pretty(&mut output, &bundle)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
output
.write_all(b"\n")
.and_then(|()| output.sync_all())
.map_err(|source| ApplicationError::WriteFile {
path: output_path.to_path_buf(),
source,
})?;
Ok(bundle)
}
/// Creates a validated online database backup.
///
/// # Errors
///
/// Returns [`ApplicationError`] when source validation or backup creation fails.
pub fn backup_database(
database_path: &Path,
backup_path: &Path,
) -> Result<BackupSummary, ApplicationError> {
SqliteStore::backup_file(database_path, backup_path)?;
Ok(BackupSummary {
database: database_path.to_string_lossy().into_owned(),
backup: backup_path.to_string_lossy().into_owned(),
})
}
/// Restores a validated backup while preserving the replaced database.
///
/// # Errors
///
/// Returns [`ApplicationError`] when locking, validation, safety backup, or restore fails.
pub fn restore_database(
database_path: &Path,
backup_path: &Path,
) -> Result<RestoreSummary, ApplicationError> {
let report = SqliteStore::restore_from(database_path, backup_path)?;
Ok(RestoreSummary {
source: report.source_path.to_string_lossy().into_owned(),
safety_backup: report
.safety_backup_path
.map(|path| path.to_string_lossy().into_owned()),
schema_version: report.schema_version,
})
}
/// Lists persisted workspaces without mutating or migrating the database.
///
/// # Errors
///
/// Returns [`ApplicationError`] when the database or registry cannot be read.
pub fn list_workspace_registry(
database_path: &Path,
) -> Result<Vec<WorkspaceRegistryItem>, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
Ok(store
.list_workspaces()?
.into_iter()
.map(|workspace| WorkspaceRegistryItem {
id: workspace.id.as_str().to_owned(),
name: workspace.name,
manifest_hash: workspace.manifest_hash,
config_path: workspace.config_path.map(|path| path.display),
repository_count: workspace.repository_count,
})
.collect())
}
/// Validates and adds a workspace manifest to the persisted registry.
///
/// # Errors
///
/// Returns [`ApplicationError`] for name mismatch, duplicate registration, invalid configuration,
/// path policy, locking, migration, or storage failure.
pub fn add_workspace_to_registry(
database_path: &Path,
workspace_name: &str,
config_path: &Path,
) -> Result<WorkspaceMutationSummary, ApplicationError> {
let context = load_workspace_context(config_path, &ScanOverrides::default())?;
if context.manifest.name != workspace_name {
return Err(ApplicationError::WorkspaceNameMismatch {
requested: workspace_name.to_owned(),
manifest: context.manifest.name,
});
}
let _lock = StoreLock::acquire(database_path, Duration::from_mins(5))?;
let mut store = SqliteStore::open(database_path)?;
if store.workspace_exists(workspace_name)? {
return Err(ApplicationError::WorkspaceAlreadyExists(
workspace_name.to_owned(),
));
}
store.save_workspace_registry(&context.registry.record)?;
Ok(WorkspaceMutationSummary {
workspace: workspace_name.to_owned(),
config_path: context.registry.record.config_path.map(|path| path.display),
repository_count: context.registry.record.repositories.len(),
operation: "add".to_owned(),
})
}
/// Removes one workspace and its snapshot state from the persisted registry.
///
/// # Errors
///
/// Returns [`ApplicationError`] for missing workspace, locking, migration, or storage failure.
pub fn remove_workspace_from_registry(
database_path: &Path,
workspace_name: &str,
) -> Result<WorkspaceMutationSummary, ApplicationError> {
let _lock = StoreLock::acquire(database_path, Duration::from_mins(5))?;
let mut store = SqliteStore::open(database_path)?;
let repository_count = store
.load_workspace_registry(workspace_name)
.map_err(|error| match error {
StoreError::RegistryIncomplete(_) => {
ApplicationError::WorkspaceNotFound(workspace_name.to_owned())
}
other => ApplicationError::Store(other),
})?
.repositories
.len();
if !store.remove_workspace(workspace_name)? {
return Err(ApplicationError::WorkspaceNotFound(
workspace_name.to_owned(),
));
}
Ok(WorkspaceMutationSummary {
workspace: workspace_name.to_owned(),
config_path: None,
repository_count,
operation: "remove".to_owned(),
})
}
/// Lists repository registrations for one persisted workspace.
///
/// # Errors
///
/// Returns [`ApplicationError`] when the database or workspace registry cannot be read.
pub fn list_repository_registry(
database_path: &Path,
workspace: &str,
) -> Result<Vec<RepositoryRecord>, ApplicationError> {
let store = SqliteStore::open_read_only(database_path)?;
Ok(store.load_workspace_registry(workspace)?.repositories)
}
/// Adds a minimal repository entry to a workspace manifest or returns a preview.
///
/// # Errors
///
/// Returns [`ApplicationError`] when parsing, path policy, backup, or atomic commit fails.
pub fn add_repository_to_manifest(
manifest_path: &Path,
alias: &str,
repository_path: &str,
dry_run: bool,
) -> Result<ManifestMutationSummary, ApplicationError> {
let source = read_file(manifest_path)?;
let edit = preview_add_repository(&source, alias, repository_path)?;
validate_manifest_edit(manifest_path, &edit)?;
finish_manifest_edit(manifest_path, edit, dry_run)
}
/// Removes one repository entry from a workspace manifest or returns a preview.
///
/// # Errors
///
/// Returns [`ApplicationError`] when parsing, path policy, backup, or atomic commit fails.
pub fn remove_repository_from_manifest(
manifest_path: &Path,
alias: &str,
dry_run: bool,
) -> Result<ManifestMutationSummary, ApplicationError> {
let source = read_file(manifest_path)?;
let edit = preview_remove_repository(&source, alias)?;
validate_manifest_edit(manifest_path, &edit)?;
finish_manifest_edit(manifest_path, edit, dry_run)
}
/// Adds one exact manual relationship declaration to a workspace manifest or returns a preview.
///
/// # Errors
///
/// Returns [`ApplicationError`] when declaration validation, path policy, backup, or atomic commit
/// fails.
pub fn add_manual_link_to_manifest(
manifest_path: &Path,
link: &ManualLinkConfig,
dry_run: bool,
) -> Result<ManifestMutationSummary, ApplicationError> {
let source = read_file(manifest_path)?;
let edit = preview_add_manual_link(&source, link)?;
validate_manifest_edit(manifest_path, &edit)?;
finish_manifest_edit(manifest_path, edit, dry_run)
}
fn validate_manifest_edit(
manifest_path: &Path,
edit: &ManifestEdit,
) -> Result<(), ApplicationError> {
let manifest = parse_manifest(&edit.updated_source)?;
register_workspace(manifest_path, &edit.updated_source, &manifest)?;
Ok(())
}
fn finish_manifest_edit(
manifest_path: &Path,
edit: ManifestEdit,
dry_run: bool,
) -> Result<ManifestMutationSummary, ApplicationError> {
if dry_run {
return Ok(ManifestMutationSummary {
summary: edit.summary,
manifest: manifest_path.to_string_lossy().into_owned(),
backup: None,
applied: false,
rendered_manifest: Some(edit.updated_source),
});
}
let report = commit_manifest_edit(manifest_path, &edit)?;
Ok(ManifestMutationSummary {
summary: edit.summary,
manifest: report.manifest_path.to_string_lossy().into_owned(),
backup: Some(report.backup_path.to_string_lossy().into_owned()),
applied: true,
rendered_manifest: None,
})
}
fn evaluate_repository_freshness(
repository: &code_system_graph_model::RepositoryRecord,
previous: Option<&RepoFreshness>,
current_manifest_hash: &str,
) -> (RepoFreshnessState, Option<String>) {
if repository.working_tree_dirty {
return (
RepoFreshnessState::WorkingTreeChanged,
Some("Git working tree has tracked or untracked changes.".to_owned()),
);
}
let Some(previous) = previous else {
return (
RepoFreshnessState::Unknown,
Some("Checkout was not observed in the published snapshot.".to_owned()),
);
};
if previous.manifest_hash != current_manifest_hash {
return (
RepoFreshnessState::ConfigChanged,
Some("Workspace manifest changed after the published snapshot.".to_owned()),
);
}
if previous.head_commit != repository.head_commit {
return (
RepoFreshnessState::CommitsBehind,
Some("Repository HEAD changed after the published snapshot.".to_owned()),
);
}
(previous.state, previous.reason.clone())
}
fn freshness_summary(repositories: &[RepoFreshness]) -> FreshnessSummary {
let overall = if repositories
.iter()
.all(|freshness| freshness.state == RepoFreshnessState::Fresh)
{
OverallFreshness::Fresh
} else if repositories.iter().any(|freshness| {
matches!(
freshness.state,
RepoFreshnessState::Unknown
| RepoFreshnessState::Unavailable
| RepoFreshnessState::Corrupt
)
}) {
OverallFreshness::Unknown
} else if repositories
.iter()
.any(|freshness| freshness.state == RepoFreshnessState::Partial)
{
OverallFreshness::Partial
} else {
OverallFreshness::Stale
};
FreshnessSummary {
overall,
stale_repositories: repositories
.iter()
.filter(|freshness| freshness.state != RepoFreshnessState::Fresh)
.map(|freshness| freshness.repo_id.clone())
.collect(),
reasons: repositories
.iter()
.filter_map(|freshness| freshness.reason.clone())
.collect(),
}
}
fn load_workspace_context(
config_path: &Path,
overrides: &ScanOverrides,
) -> Result<WorkspaceContext, ApplicationError> {
let manifest_source = read_file(config_path)?;
let manifest = parse_manifest(&manifest_source)?;
validate_global_policy_source(config_path, &manifest)?;
let extraction_budgets = ExtractionBudgets::resolve(manifest.extraction_budgets.as_ref())
.map_err(ManifestError::from)?;
let execution_policy = ExecutionPolicy::resolve(manifest.execution_policy.as_ref())
.map_err(ManifestError::from)?;
for alias in overrides.repo_openapi.keys() {
if !manifest.repos.contains_key(alias) {
return Err(ApplicationError::UnknownOverrideRepository(alias.clone()));
}
}
let mut registry = register_workspace(config_path, &manifest_source, &manifest)?;
let mut repository_configs = BTreeMap::new();
let mut semantic_manifest = manifest.clone();
semantic_manifest.execution_policy = None;
let mut fingerprint_material = serde_json::to_string(&semantic_manifest)
.map_err(|error| ApplicationError::Initialization(error.to_string()))?;
for (alias, repository) in &manifest.repos {
let checkout_path = registry
.checkout_path(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let mut effective = resolve_repository_config(checkout_path, repository)?;
if let Some(openapi) = overrides.repo_openapi.get(alias) {
apply_openapi_override(&mut effective, openapi)?;
}
fingerprint_material.push('\n');
fingerprint_material.push_str(alias);
fingerprint_material.push(':');
fingerprint_material.push_str(&effective.fingerprint);
repository_configs.insert(alias.clone(), effective);
}
registry.record.manifest_hash = stable_id("manifest", &fingerprint_material);
Ok(WorkspaceContext {
manifest,
registry,
repository_configs,
extraction_budgets,
execution_policy,
})
}
fn validate_global_policy_source(
config_path: &Path,
manifest: &WorkspaceManifest,
) -> Result<(), ApplicationError> {
if manifest.extraction_budgets.is_none() && manifest.execution_policy.is_none() {
return Ok(());
}
let canonical_config =
fs::canonicalize(config_path).map_err(|source| ApplicationError::ReadFile {
path: config_path.to_path_buf(),
source,
})?;
let base = canonical_config.parent().ok_or_else(|| {
ApplicationError::Initialization(format!(
"workspace manifest `{}` has no parent directory",
canonical_config.display()
))
})?;
for (alias, repository) in &manifest.repos {
let configured = Path::new(&repository.path);
let candidate = if configured.is_absolute() {
configured.to_path_buf()
} else {
base.join(configured)
};
let checkout =
fs::canonicalize(&candidate).map_err(|source| ApplicationError::ReadFile {
path: candidate.clone(),
source,
})?;
if canonical_config.starts_with(&checkout) {
return Err(ApplicationError::UntrustedGlobalPolicySource {
config: canonical_config,
repository: alias.clone(),
});
}
}
Ok(())
}
fn focused_batch_cache_complete(
fingerprints: &[ArtifactFingerprint],
stored: &[StoredExtractorBatch],
budgets: &ExtractionBudgets,
) -> bool {
let budget_fingerprint = budgets.fingerprint();
fingerprints
.iter()
.filter(|fingerprint| focused_extractor(&fingerprint.extractor))
.all(|fingerprint| {
stored.iter().any(|batch| {
batch.source == *fingerprint
&& batch.extractor_version == EXTRACTION_CONTRACT_VERSION
&& batch.budget_fingerprint == budget_fingerprint
})
})
}
fn stored_batch_degradations(
stored: &[StoredExtractorBatch],
budgets: &ExtractionBudgets,
) -> Result<Vec<String>, ApplicationError> {
let mut degradations = Vec::new();
for batch in stored {
if batch.source_was_lossy {
degradations.push(format!("{} contains invalid UTF-8 and was decoded lossily; extracted evidence is incomplete", batch.source.path.display));
}
if batch.source.extractor == "code-system-graph.data.artifact" {
let decoded: ExtractorBatch<DataDocument> =
load_extractor_batch_with_budgets(batch, budgets)?;
for document in decoded.outputs {
if document.incomplete {
degradations.push(format!(
"{} data extraction is incomplete: {:?}",
batch.source.path.display, document.warnings
));
}
}
}
}
Ok(degradations)
}
fn finalize_scan_degradations(mut degradations: Vec<String>) -> (usize, Vec<String>) {
degradations.sort();
degradations.dedup();
let count = degradations.len();
if count > MAX_SCAN_DEGRADATIONS {
degradations.truncate(MAX_SCAN_DEGRADATIONS - 1);
degradations.push(format!(
"{} additional degradations omitted",
count - (MAX_SCAN_DEGRADATIONS - 1)
));
}
(count, degradations)
}
#[expect(
clippy::too_many_lines,
reason = "Batch decode, reuse, extraction, and persistence share one fail-closed type boundary"
)]
fn assemble_focused_batches(
context: &WorkspaceContext,
fingerprints: &[ArtifactFingerprint],
previous: &[StoredExtractorBatch],
plan: &ExtractorBatchPlan,
cached_keys: &BTreeSet<ArtifactKey>,
work_state: &mut work_state::WorkState,
) -> Result<FocusedBatchState, ApplicationError> {
let previous_by_key = previous
.iter()
.map(|batch| (ArtifactKey::from(&batch.source), batch))
.collect::<BTreeMap<_, _>>();
let planned_actions = plan
.batches
.iter()
.map(|batch| (batch.key.clone(), batch.action))
.collect::<BTreeMap<_, _>>();
let mut previous_source_batches = previous
.iter()
.filter(|batch| source_extractor(&batch.source.extractor))
.map(|batch| load_extractor_batch_with_budgets(batch, &context.extraction_budgets))
.collect::<Result<Vec<ExtractorBatch<SourceObservation>>, _>>()?;
previous_source_batches.sort_by_key(ExtractorBatch::key);
let checkouts = context
.registry
.record
.repositories
.iter()
.filter_map(|repository| {
context
.registry
.checkout_path(&repository.alias)
.map(|path| (repository.id.clone(), path))
})
.collect::<BTreeMap<_, _>>();
let mut source_batches = Vec::new();
let mut package_batches = Vec::new();
let mut generated_client_batches = Vec::new();
let mut graphql_batches = Vec::new();
let mut event_batches = Vec::new();
let mut protobuf_batches = Vec::new();
let mut data_batches = Vec::new();
let mut infrastructure_batches = Vec::new();
let mut documentation_batches = Vec::new();
let mut config_batches = Vec::new();
let mut stored_batches = Vec::new();
let mut degradations = Vec::new();
let mut force_relink = false;
let mut checkpoint_writes = 0_u64;
let mut artifact_durations_ms = Vec::new();
for fingerprint in fingerprints
.iter()
.filter(|fingerprint| focused_extractor(&fingerprint.extractor))
{
let artifact_started = Instant::now();
let key = ArtifactKey::from(fingerprint);
let reusable = previous_by_key.get(&key).copied().filter(|batch| {
(planned_actions.get(&key) == Some(&BatchAction::Reuse) || cached_keys.contains(&key))
&& batch.source.content_hash == fingerprint.content_hash
&& batch.extractor_version == EXTRACTION_CONTRACT_VERSION
&& batch.budget_fingerprint == context.extraction_budgets.fingerprint()
});
if let Some(stored) = reusable {
stored_batches.push(stored.clone());
if source_extractor(&fingerprint.extractor) {
source_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if fingerprint.extractor == "code-system-graph.packages" {
package_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if graphql_extractor(&fingerprint.extractor) {
graphql_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if event_extractor(&fingerprint.extractor) {
event_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if protobuf_extractor(&fingerprint.extractor) {
protobuf_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if data_extractor(&fingerprint.extractor) {
data_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if infrastructure_extractor(&fingerprint.extractor) {
infrastructure_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if documentation_extractor(&fingerprint.extractor) {
documentation_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else if fingerprint.extractor == "code-system-graph.config.safe" {
config_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
} else {
generated_client_batches.push(load_extractor_batch_with_budgets(
stored,
&context.extraction_budgets,
)?);
}
worker::report_progress(code_system_graph_core::JobPhase::Extraction, 1);
artifact_durations_ms.push(duration_millis(artifact_started.elapsed()));
continue;
}
if previous_by_key.contains_key(&key) {
force_relink = true;
}
let checkout = checkouts.get(&fingerprint.repo_id).ok_or_else(|| {
ApplicationError::RegistryAliasMissing(fingerprint.repo_id.as_str().to_owned())
})?;
let relative_path = native_relative_path(&fingerprint.path);
let artifact_path = checkout.join(&relative_path);
let mut tracker = ExtractionTracker::new(
&fingerprint.path.display,
&fingerprint.extractor,
&context.extraction_budgets,
);
let (source, source_was_lossy) = read_source_file(&artifact_path, &mut tracker)?;
if source_was_lossy {
degradations.push(format!("{} contains invalid UTF-8 and was decoded lossily; extracted evidence is incomplete", fingerprint.path.display));
}
let mut persist = |stored: StoredExtractorBatch| -> Result<(), ApplicationError> {
if work_state
.put_batch(
&stored,
context.execution_policy.max_checkpoint_cache_bytes,
current_unix_millis(),
)
.map_err(ApplicationError::Initialization)?
{
checkpoint_writes = checkpoint_writes.checked_add(1).ok_or_else(|| {
ApplicationError::Initialization(
"checkpoint write counter overflowed".to_owned(),
)
})?;
}
stored_batches.push(stored);
Ok(())
};
if source_extractor(&fingerprint.extractor) {
let syntax = inspect_source_syntax(
source_syntax_language(&fingerprint.extractor),
&portable_path(&fingerprint.path.display),
&source,
&mut tracker,
)?;
let reserved_observations =
u64::try_from(syntax.boundary_candidate_count).map_err(|_| {
ApplicationError::InvalidSourceObservation(format!(
"{} contains too many syntax candidates",
fingerprint.path.display
))
})?;
tracker.charge_work(reserved_observations)?;
precheck_focused_source_values(&source, &mut tracker)?;
let mut observations = match fingerprint.extractor.as_str() {
"code-system-graph.source.javascript" => {
parse_javascript_source_at_path_with_tracker(
&portable_path(&fingerprint.path.display),
&source,
&mut tracker,
)?
}
"code-system-graph.source.typescript" => {
parse_typescript_source_at_path_with_tracker(
&portable_path(&fingerprint.path.display),
&source,
&mut tracker,
)?
}
"code-system-graph.source.rust" => {
parse_rust_source_with_tracker(&source, &mut tracker)?
}
"code-system-graph.source.python" => {
parse_python_source_with_tracker(&source, &mut tracker)?
}
"code-system-graph.source.go" => {
parse_go_source_with_tracker(&source, &mut tracker)?
}
"code-system-graph.source.java" => {
parse_java_source_with_tracker(&source, &mut tracker)?
}
_ => Vec::new(),
};
if observations
.iter()
.any(|observation| observation.role != SourceRole::Test)
&& syntax.boundary_candidate_count == 0
{
return Err(ApplicationError::InvalidSourceObservation(format!(
"{} produced framework facts without a Tree-sitter boundary candidate",
fingerprint.path.display
)));
}
if syntax.has_error {
for observation in &mut observations {
observation.status = SourceEpistemicStatus::Incomplete;
if !observation
.warnings
.contains(&SourceWarning::SyntaxErrorRecovery)
{
observation
.warnings
.push(SourceWarning::SyntaxErrorRecovery);
}
}
}
let batch = ExtractorBatch::new(fingerprint.clone(), observations);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
source_batches.push(batch);
} else if fingerprint.extractor == "code-system-graph.packages" {
let portable_path = portable_path(&fingerprint.path.display);
let manifest =
extract_package_manifest_with_tracker(&portable_path, &source, &mut tracker)?;
let batch = ExtractorBatch::new(fingerprint.clone(), vec![manifest]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
package_batches.push(batch);
} else if graphql_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = match fingerprint.extractor.as_str() {
"code-system-graph.graphql.document" => {
extract_graphql_document_with_tracker(&portable_path, &source, &mut tracker)?
}
"code-system-graph.graphql.persisted" => GraphqlDocument {
source_path: portable_path.clone(),
types: Vec::new(),
operations: Vec::new(),
fragments: Vec::new(),
persisted_operations: extract_graphql_persisted_operations_with_tracker(
&portable_path,
&source,
&mut tracker,
)?,
resolvers: Vec::new(),
federation: Vec::new(),
complete: true,
warnings: Vec::new(),
},
"code-system-graph.graphql.source" => {
let language = source_language_for_path(&relative_path).ok_or_else(|| {
ApplicationError::InvalidSourceObservation(format!(
"unsupported GraphQL source language for `{portable_path}`"
))
})?;
let mut document =
parse_graphql_source_with_tracker(language, &source, &mut tracker)?;
document.source_path = portable_path;
document
}
_ => unreachable!("graphql extractor classification must be exhaustive"),
};
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
graphql_batches.push(batch);
} else if event_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = if fingerprint.extractor == "code-system-graph.events.asyncapi" {
extract_asyncapi(&portable_path, &source)?
} else {
let language = source_language_for_path(&relative_path).ok_or_else(|| {
ApplicationError::InvalidSourceObservation(format!(
"unsupported event source language for `{portable_path}`"
))
})?;
let mut document = parse_event_source(language, &source);
document.source_path = Some(portable_path);
document
};
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
event_batches.push(batch);
} else if protobuf_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = if fingerprint.extractor == "code-system-graph.protobuf" {
ProtobufDocument::File(Box::new(extract_protobuf_with_tracker(
&portable_path,
&source,
&mut tracker,
)?))
} else {
let language = source_language_for_path(&relative_path).ok_or_else(|| {
ApplicationError::InvalidSourceObservation(format!(
"unsupported generated protobuf source language for `{portable_path}`"
))
})?;
ProtobufDocument::Generated(parse_protobuf_generated_source(
language,
&portable_path,
&source,
))
};
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
protobuf_batches.push(batch);
} else if data_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = if fingerprint.extractor == "code-system-graph.data.source" {
let language = source_language_for_path(&relative_path).ok_or_else(|| {
ApplicationError::InvalidSourceObservation(format!(
"unsupported data source language for `{portable_path}`"
))
})?;
let crate_root = cargo_crate_root(checkout, &artifact_path);
parse_literal_sql_source_at_root(language, &portable_path, &crate_root, &source)
} else {
extract_data_artifact(&portable_path, &source)?
};
if fingerprint.extractor == "code-system-graph.data.artifact" && document.incomplete {
degradations.push(format!(
"{} data extraction is incomplete: {:?}",
fingerprint.path.display, document.warnings
));
}
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
data_batches.push(batch);
} else if infrastructure_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = match fingerprint.extractor.as_str() {
"code-system-graph.infrastructure.compose" => {
extract_docker_compose(&portable_path, &source)?
}
"code-system-graph.infrastructure.kubernetes" => {
extract_kubernetes(&portable_path, &source)?
}
"code-system-graph.infrastructure.helm" => extract_helm(&portable_path, &source)?,
"code-system-graph.infrastructure.terraform" => {
extract_terraform(&portable_path, &source)?
}
_ => unreachable!("infrastructure extractor classification must be exhaustive"),
};
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
infrastructure_batches.push(batch);
} else if documentation_extractor(&fingerprint.extractor) {
let portable_path = portable_path(&fingerprint.path.display);
let document = match fingerprint.extractor.as_str() {
"code-system-graph.documents.markdown" => {
extract_markdown(&portable_path, &source)?
}
"code-system-graph.documents.codeowners" => {
extract_codeowners(&portable_path, &source)?
}
"code-system-graph.documents.catalog" => {
extract_service_catalog(&portable_path, &source)?
}
_ => unreachable!("documentation extractor classification must be exhaustive"),
};
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
documentation_batches.push(batch);
} else if fingerprint.extractor == "code-system-graph.config.safe" {
let portable_path = portable_path(&fingerprint.path.display);
let document = extract_safe_config(&portable_path, &source)?;
let batch = ExtractorBatch::new(fingerprint.clone(), vec![document]);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
config_batches.push(batch);
} else {
let portable_path = portable_path(&fingerprint.path.display);
let metadata =
extract_generated_client_metadata(&portable_path, &source, &mut tracker)?;
let batch = ExtractorBatch::new(fingerprint.clone(), metadata);
persist(store_extractor_batch(
&batch,
&mut tracker,
source_was_lossy,
)?)?;
generated_client_batches.push(batch);
}
worker::report_progress(code_system_graph_core::JobPhase::Extraction, 1);
artifact_durations_ms.push(duration_millis(artifact_started.elapsed()));
}
source_batches.sort_by_key(ExtractorBatch::key);
package_batches.sort_by_key(ExtractorBatch::key);
generated_client_batches.sort_by_key(ExtractorBatch::key);
graphql_batches.sort_by_key(ExtractorBatch::key);
event_batches.sort_by_key(ExtractorBatch::key);
protobuf_batches.sort_by_key(ExtractorBatch::key);
data_batches.sort_by_key(ExtractorBatch::key);
infrastructure_batches.sort_by_key(ExtractorBatch::key);
documentation_batches.sort_by_key(ExtractorBatch::key);
config_batches.sort_by_key(ExtractorBatch::key);
stored_batches.sort_by(|left, right| {
ArtifactKey::from(&left.source).cmp(&ArtifactKey::from(&right.source))
});
Ok(FocusedBatchState {
source_batches,
previous_source_batches,
package_batches,
generated_client_batches,
graphql_batches,
event_batches,
protobuf_batches,
data_batches,
infrastructure_batches,
documentation_batches,
config_batches,
stored_batches,
degradations,
force_relink,
checkpoint_writes,
artifact_durations_ms,
})
}
fn focused_extractor(extractor: &str) -> bool {
source_extractor(extractor)
|| matches!(
extractor,
"code-system-graph.packages" | "code-system-graph.http.generated-client"
)
|| graphql_extractor(extractor)
|| event_extractor(extractor)
|| protobuf_extractor(extractor)
|| data_extractor(extractor)
|| infrastructure_extractor(extractor)
|| documentation_extractor(extractor)
|| extractor == "code-system-graph.config.safe"
}
fn graphql_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.graphql.document"
| "code-system-graph.graphql.persisted"
| "code-system-graph.graphql.source"
)
}
fn event_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.events.asyncapi" | "code-system-graph.events.source"
)
}
fn protobuf_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.protobuf" | "code-system-graph.protobuf.generated"
)
}
fn data_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.data.artifact" | "code-system-graph.data.source"
)
}
fn infrastructure_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.infrastructure.compose"
| "code-system-graph.infrastructure.kubernetes"
| "code-system-graph.infrastructure.helm"
| "code-system-graph.infrastructure.terraform"
)
}
fn documentation_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.documents.markdown"
| "code-system-graph.documents.codeowners"
| "code-system-graph.documents.catalog"
)
}
fn source_extractor(extractor: &str) -> bool {
matches!(
extractor,
"code-system-graph.source.javascript"
| "code-system-graph.source.typescript"
| "code-system-graph.source.rust"
| "code-system-graph.source.python"
| "code-system-graph.source.go"
| "code-system-graph.source.java"
)
}
fn source_syntax_language(extractor: &str) -> SourceSyntaxLanguage {
match extractor {
"code-system-graph.source.javascript" => SourceSyntaxLanguage::JavaScript,
"code-system-graph.source.typescript" => SourceSyntaxLanguage::TypeScript,
"code-system-graph.source.rust" => SourceSyntaxLanguage::Rust,
"code-system-graph.source.python" => SourceSyntaxLanguage::Python,
"code-system-graph.source.go" => SourceSyntaxLanguage::Go,
"code-system-graph.source.java" => SourceSyntaxLanguage::Java,
_ => unreachable!("source_syntax_language requires a focused source extractor"),
}
}
fn source_language_for_path(path: &Path) -> Option<SourceLanguage> {
let extension = path.extension()?.to_str()?;
match extension {
"js" | "jsx" => Some(SourceLanguage::JavaScript),
"ts" | "tsx" => Some(SourceLanguage::TypeScript),
"rs" => Some(SourceLanguage::Rust),
"py" => Some(SourceLanguage::Python),
"go" => Some(SourceLanguage::Go),
"java" => Some(SourceLanguage::Java),
_ => None,
}
}
#[cfg(unix)]
fn native_relative_path(path: &code_system_graph_model::NativePath) -> PathBuf {
use std::os::unix::ffi::OsStringExt;
PathBuf::from(std::ffi::OsString::from_vec(path.bytes.clone()))
}
#[cfg(windows)]
fn native_relative_path(path: &code_system_graph_model::NativePath) -> PathBuf {
use std::os::windows::ffi::OsStringExt;
let wide = path
.bytes
.chunks_exact(2)
.map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
.collect::<Vec<_>>();
PathBuf::from(std::ffi::OsString::from_wide(&wide))
}
#[cfg(not(any(unix, windows)))]
fn native_relative_path(path: &code_system_graph_model::NativePath) -> PathBuf {
PathBuf::from(String::from_utf8_lossy(&path.bytes).into_owned())
}
fn generated_client_graph(
batches: &[ExtractorBatch<GeneratedClientMetadata>],
boundaries: &[HttpBoundary],
) -> (Vec<Node>, Vec<Edge>, Vec<Evidence>) {
let mut nodes = Vec::new();
let mut edges = Vec::new();
let mut evidence = Vec::new();
for batch in batches {
let source_path = portable_path(&batch.source.path.display);
for metadata in &batch.outputs {
let name = metadata.name.as_deref().unwrap_or("metadata");
let stable_key = format!(
"generated-client:{}:{source_path}:{name}:{}",
batch.source.repo_id.as_str(),
metadata.generated_file.as_deref().unwrap_or("")
);
let node = Node {
id: NodeId::new(stable_id("node", &stable_key)),
kind: NodeKind::Artifact,
repo_id: Some(batch.source.repo_id.clone()),
stable_key: stable_key.clone(),
label: metadata
.generator_name
.as_ref()
.map_or_else(|| format!("{} {name}", metadata.tool), Clone::clone),
};
let item_evidence = Evidence {
id: EvidenceId::new(stable_id("evidence", &stable_key)),
repo_id: Some(batch.source.repo_id.clone()),
file_path: Some(source_path.clone()),
start_line: Some(metadata.line),
end_line: Some(metadata.line),
extractor: "code-system-graph.http.generated-client".to_owned(),
extractor_version: "1.0.0".to_owned(),
provenance: Provenance::Extracted,
confidence: 1.0,
observed_at_commit: None,
content_hash: Some(batch.source.content_hash.clone()),
note: Some("explicit generated-client metadata".to_owned()),
};
if let Some(input_spec) = &metadata.input_spec {
edges.extend(
boundaries
.iter()
.filter(|boundary| {
boundary.node.repo_id.as_ref() == Some(&batch.source.repo_id)
&& boundary.evidence.file_path.as_ref() == Some(input_spec)
})
.map(|boundary| {
let key = format!(
"{}:{:?}:{}",
node.id.as_str(),
EdgeKind::Consumes,
boundary.node.id.as_str()
);
Edge {
id: EdgeId::new(stable_id("edge", &key)),
source: node.id.clone(),
target: boundary.node.id.clone(),
kind: EdgeKind::Consumes,
confidence: 1.0,
status: EpistemicStatus::Confirmed,
evidence: vec![item_evidence.id.clone()],
}
}),
);
}
nodes.push(node);
evidence.push(item_evidence);
}
}
nodes.sort_by(|left, right| left.id.cmp(&right.id));
nodes.dedup_by(|left, right| left.id == right.id);
edges.sort_by(|left, right| left.id.cmp(&right.id));
edges.dedup_by(|left, right| left.id == right.id);
evidence.sort_by(|left, right| left.id.cmp(&right.id));
evidence.dedup_by(|left, right| left.id == right.id);
(nodes, edges, evidence)
}
#[expect(
clippy::too_many_lines,
reason = "Graph assembly keeps every source family in one deterministic merge boundary"
)]
fn assemble_graph(
context: &WorkspaceContext,
fingerprints: &[ArtifactFingerprint],
focused: &FocusedBatchState,
) -> Result<GraphAssembly, ApplicationError> {
let mut boundaries = extract_boundaries(context)?;
let mut tests = extract_declared_tests(context, fingerprints)?;
let mut implementations = extract_declared_implementations(context, fingerprints)?;
let source_facts = focused
.source_batches
.iter()
.map(|batch| {
source_observations_to_graph(
&batch.source.repo_id,
&portable_path(&batch.source.path.display),
&batch.source.content_hash,
&batch.outputs,
)
})
.collect::<Vec<SourceGraphFacts>>();
for facts in &source_facts {
boundaries.extend(facts.boundaries.clone());
tests.extend(facts.tests.clone());
implementations.extend(facts.implementations.clone());
}
let mut package_facts = focused
.package_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|manifest| {
package_manifest_to_graph(
&batch.source.repo_id,
&portable_path(&batch.source.path.display),
&batch.source.content_hash,
manifest,
)
})
})
.collect::<Vec<PackageGraphFacts>>();
link_registered_package_owners(&mut package_facts);
let (generated_nodes, generated_edges, generated_evidence) =
generated_client_graph(&focused.generated_client_batches, &boundaries);
let graphql_sources = focused
.graphql_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let graphql_inputs = graphql_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let GraphqlGraphFacts {
nodes: graphql_nodes,
edges: graphql_edges,
evidence: graphql_evidence,
} = graphql_documents_to_graph(&graphql_inputs);
let event_sources = focused
.event_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let event_inputs = event_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let EventGraphFacts {
nodes: event_nodes,
edges: event_edges,
evidence: event_evidence,
} = event_documents_to_graph(&event_inputs);
let protobuf_sources = focused
.protobuf_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let protobuf_inputs = protobuf_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let ProtobufGraphFacts {
nodes: protobuf_nodes,
edges: protobuf_edges,
evidence: protobuf_evidence,
} = protobuf_documents_to_graph(&protobuf_inputs);
let data_sources = focused
.data_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let data_inputs = data_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let infrastructure_sources = focused
.infrastructure_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let infrastructure_inputs = infrastructure_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let documentation_sources = focused
.documentation_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let documentation_inputs = documentation_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let config_sources = focused
.config_batches
.iter()
.flat_map(|batch| {
batch.outputs.iter().map(|document| {
(
batch.source.repo_id.clone(),
portable_path(&batch.source.path.display),
batch.source.content_hash.clone(),
document,
)
})
})
.collect::<Vec<_>>();
let config_inputs = config_sources
.iter()
.map(|(repo_id, path, hash, document)| (repo_id, path.as_str(), hash.as_str(), *document))
.collect::<Vec<_>>();
let mut known_nodes = boundaries
.iter()
.map(|boundary| boundary.node.clone())
.chain(tests.iter().map(|test| test.node.clone()))
.chain(
implementations
.iter()
.map(|implementation| implementation.node.clone()),
)
.chain(
source_facts
.iter()
.flat_map(|facts| facts.standalone_test_nodes.iter().cloned()),
)
.chain(
source_facts
.iter()
.flat_map(|facts| facts.relation_nodes.iter().cloned()),
)
.chain(
package_facts
.iter()
.flat_map(|facts| facts.nodes.iter().cloned()),
)
.chain(generated_nodes.iter().cloned())
.chain(graphql_nodes.iter().cloned())
.chain(event_nodes.iter().cloned())
.chain(protobuf_nodes.iter().cloned())
.collect::<Vec<_>>();
known_nodes.sort_by(|left, right| left.id.cmp(&right.id));
known_nodes.dedup_by(|left, right| left.id == right.id);
let repository_aliases = context
.registry
.record
.repositories
.iter()
.map(|repository| (repository.alias.as_str(), &repository.id))
.collect::<Vec<_>>();
let ExtractionGraphFacts {
nodes: extraction_nodes,
edges: extraction_edges,
evidence: extraction_evidence,
} = documents_to_graph(
&data_inputs,
&infrastructure_inputs,
&documentation_inputs,
&config_inputs,
&known_nodes,
&repository_aliases,
);
let mut http_links = link_http_boundaries_with_ambiguities(&boundaries);
let test_links = link_declared_tests_with_ambiguities(&tests, &boundaries);
let implementation_links =
link_declared_implementations_with_ambiguities(&implementations, &boundaries);
http_links.ambiguities.extend(test_links.ambiguities);
http_links
.ambiguities
.extend(implementation_links.ambiguities);
http_links.ambiguities.sort_by(|left, right| {
(&left.method, &left.path, &left.candidates).cmp(&(
&right.method,
&right.path,
&right.candidates,
))
});
http_links.ambiguities.dedup();
let degradations = http_links
.ambiguities
.iter()
.map(|ambiguity| {
format!(
"ambiguous HTTP provider for {} {}; stable candidates: {}",
ambiguity.method,
ambiguity.path,
ambiguity.candidates.join(", ")
)
})
.collect::<Vec<_>>();
let mut edges = http_links.edges;
edges.extend(test_links.edges);
edges.extend(implementation_links.edges);
edges.extend(
source_facts
.iter()
.flat_map(|facts| facts.relation_edges.iter().cloned()),
);
edges.extend(
package_facts
.iter()
.flat_map(|facts| facts.edges.iter().cloned()),
);
edges.extend(generated_edges);
edges.extend(graphql_edges);
edges.extend(event_edges);
edges.extend(protobuf_edges);
edges.extend(extraction_edges);
edges.sort_by(|left, right| left.id.cmp(&right.id));
edges.dedup_by(|left, right| left.id == right.id);
let mut nodes = boundaries
.iter()
.map(|boundary| (boundary.node.id.clone(), boundary.node.clone()))
.collect::<BTreeMap<_, _>>();
nodes.extend(
tests
.iter()
.map(|test| (test.node.id.clone(), test.node.clone())),
);
nodes.extend(
implementations
.iter()
.map(|implementation| (implementation.node.id.clone(), implementation.node.clone())),
);
nodes.extend(source_facts.iter().flat_map(|facts| {
facts
.standalone_test_nodes
.iter()
.map(|node| (node.id.clone(), node.clone()))
}));
nodes.extend(source_facts.iter().flat_map(|facts| {
facts
.relation_nodes
.iter()
.map(|node| (node.id.clone(), node.clone()))
}));
nodes.extend(package_facts.iter().flat_map(|facts| {
facts
.nodes
.iter()
.map(|node| (node.id.clone(), node.clone()))
}));
nodes.extend(
generated_nodes
.into_iter()
.map(|node| (node.id.clone(), node)),
);
nodes.extend(
graphql_nodes
.into_iter()
.map(|node| (node.id.clone(), node)),
);
nodes.extend(event_nodes.into_iter().map(|node| (node.id.clone(), node)));
nodes.extend(
protobuf_nodes
.into_iter()
.map(|node| (node.id.clone(), node)),
);
nodes.extend(
extraction_nodes
.into_iter()
.map(|node| (node.id.clone(), node)),
);
let mut evidence = boundaries
.iter()
.map(|boundary| (boundary.evidence.id.clone(), boundary.evidence.clone()))
.collect::<BTreeMap<_, _>>();
evidence.extend(
tests
.iter()
.map(|test| (test.evidence.id.clone(), test.evidence.clone())),
);
evidence.extend(implementations.iter().map(|implementation| {
(
implementation.evidence.id.clone(),
implementation.evidence.clone(),
)
}));
evidence.extend(source_facts.iter().flat_map(|facts| {
facts
.standalone_test_evidence
.iter()
.map(|evidence| (evidence.id.clone(), evidence.clone()))
}));
evidence.extend(source_facts.iter().flat_map(|facts| {
facts
.relation_evidence
.iter()
.map(|evidence| (evidence.id.clone(), evidence.clone()))
}));
evidence.extend(package_facts.iter().flat_map(|facts| {
facts
.evidence
.iter()
.map(|evidence| (evidence.id.clone(), evidence.clone()))
}));
evidence.extend(
generated_evidence
.into_iter()
.map(|item| (item.id.clone(), item)),
);
evidence.extend(
graphql_evidence
.into_iter()
.map(|item| (item.id.clone(), item)),
);
evidence.extend(
event_evidence
.into_iter()
.map(|item| (item.id.clone(), item)),
);
evidence.extend(
protobuf_evidence
.into_iter()
.map(|item| (item.id.clone(), item)),
);
evidence.extend(
extraction_evidence
.into_iter()
.map(|item| (item.id.clone(), item)),
);
let mut link_node_keys = BTreeMap::new();
link_node_keys.extend(boundaries.iter().map(|boundary| {
(
boundary.node.id.clone(),
format!("{}:{}", boundary.method, boundary.path),
)
}));
link_node_keys.extend(tests.iter().map(|test| {
(
test.node.id.clone(),
format!("{}:{}", test.method, test.path),
)
}));
link_node_keys.extend(implementations.iter().map(|implementation| {
(
implementation.node.id.clone(),
format!("{}:{}", implementation.method, implementation.path),
)
}));
Ok(GraphAssembly {
nodes: nodes.into_values().collect(),
edges,
evidence: evidence.into_values().collect(),
link_decisions: Vec::new(),
link_node_keys,
degradations,
})
}
#[expect(
clippy::too_many_lines,
reason = "Fail-closed relinking keeps old/new neighborhood handling in one audit boundary"
)]
fn relink_affected_graph(
graph: &mut GraphAssembly,
plan: &IncrementalPlan,
batch_plan: &ExtractorBatchPlan,
focused: &FocusedBatchState,
_previous_nodes: &[Node],
previous_edges: &[Edge],
) -> Result<(), ApplicationError> {
let full_relink = focused.force_relink
|| plan.changes.iter().any(|change| {
change.kind != code_system_graph_model::ArtifactChangeKind::Unchanged
&& change.extractor != "code-system-graph.packages"
&& !source_extractor(&change.extractor)
});
if full_relink || previous_edges.is_empty() {
return Ok(());
}
let source_plan = ExtractorBatchPlan {
batches: batch_plan
.batches
.iter()
.filter(|batch| source_extractor(&batch.key.extractor))
.cloned()
.collect(),
};
let affected = affected_link_keys(
&source_plan,
&focused.previous_source_batches,
&focused.source_batches,
|observation| {
observation
.method
.as_ref()
.zip(observation.path.as_ref())
.map(|(method, path)| format!("{method}:{path}"))
},
)?
.into_iter()
.flatten()
.collect::<BTreeSet<_>>();
if affected.is_empty() {
return Ok(());
}
let mut previous_node_keys = graph.link_node_keys.clone();
for batch in &focused.previous_source_batches {
let facts = source_observations_to_graph(
&batch.source.repo_id,
&portable_path(&batch.source.path.display),
&batch.source.content_hash,
&batch.outputs,
);
previous_node_keys.extend(facts.boundaries.iter().map(|boundary| {
(
boundary.node.id.clone(),
format!("{}:{}", boundary.method, boundary.path),
)
}));
previous_node_keys.extend(facts.tests.iter().map(|test| {
(
test.node.id.clone(),
format!("{}:{}", test.method, test.path),
)
}));
previous_node_keys.extend(facts.implementations.iter().map(|implementation| {
(
implementation.node.id.clone(),
format!("{}:{}", implementation.method, implementation.path),
)
}));
}
let current_node_ids = graph
.nodes
.iter()
.map(|node| node.id.clone())
.collect::<BTreeSet<_>>();
let is_http_link = |edge: &Edge| {
matches!(
edge.kind,
code_system_graph_model::EdgeKind::CallsRemote
| code_system_graph_model::EdgeKind::Validates
| code_system_graph_model::EdgeKind::ImplementedBy
)
};
let recomputed_http = graph
.edges
.iter()
.filter(|edge| is_http_link(edge))
.cloned()
.collect::<Vec<_>>();
let previous_http = previous_edges
.iter()
.filter(|edge| is_http_link(edge))
.cloned()
.collect::<Vec<_>>();
let mut edges = graph
.edges
.iter()
.filter(|edge| !is_http_link(edge))
.cloned()
.collect::<Vec<_>>();
edges.extend(merge_affected_link_neighborhoods(
&previous_http,
&recomputed_http,
&affected,
&previous_node_keys,
&graph.link_node_keys,
¤t_node_ids,
));
edges.sort_by(|left, right| left.id.cmp(&right.id));
edges.dedup_by(|left, right| left.id == right.id);
graph.edges = edges;
Ok(())
}
fn run_codegraph_corroboration(
context: &WorkspaceContext,
focused: &FocusedBatchState,
plan: &IncrementalPlan,
selected_repository: Option<&str>,
codegraph_binary: Option<PathBuf>,
) -> CorroborationSummary {
let (jobs, setup_degradations) =
prepare_codegraph_jobs(context, focused, plan, selected_repository);
let worker = std::thread::spawn(move || -> Result<Vec<RepositoryCorroboration>, String> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| format!("cannot start CodeGraph corroboration runtime: {error}"))?;
runtime.block_on(async move {
let mut config = CodeGraphConfig::default();
if let Some(binary) = codegraph_binary {
config.binary = binary.into_os_string();
}
let provider = CodeGraphProvider::new(config).map_err(|error| error.to_string())?;
let mut reports = Vec::new();
for (repo_id, project_path, anchors, changed_files) in jobs {
let report = corroborate_repository(
&provider,
repo_id.clone(),
project_path,
&anchors,
&changed_files,
ProviderBudget {
timeout: Duration::from_secs(2),
max_output_bytes: 256 * 1024,
max_items: 20,
},
CancellationToken::new(),
)
.await;
reports.push(RepositoryCorroboration { repo_id, report });
}
provider
.shutdown()
.await
.map_err(|error| error.to_string())?;
Ok(reports)
})
});
let mut summary = CorroborationSummary {
degradations: setup_degradations,
..CorroborationSummary::default()
};
match worker.join() {
Ok(Ok(reports)) => {
summary.confirmed_symbols = reports
.iter()
.flat_map(|item| &item.report.symbols)
.filter(|symbol| matches!(symbol, SymbolCorroboration::Confirmed { .. }))
.count();
summary.affected_tests = reports
.iter()
.map(|item| item.report.affected_tests.len())
.sum();
summary.degradations.extend(
reports
.iter()
.flat_map(|item| item.report.degradations.iter().cloned()),
);
summary.reports = reports;
}
Ok(Err(error)) => summary.degradations.push(error),
Err(_) => summary
.degradations
.push("CodeGraph corroboration worker terminated unexpectedly".to_owned()),
}
summary.degradations.sort();
summary.degradations.dedup();
summary
}
type CodeGraphJob = (RepoId, PathBuf, Vec<SymbolAnchor>, Vec<String>);
fn prepare_codegraph_jobs(
context: &WorkspaceContext,
focused: &FocusedBatchState,
plan: &IncrementalPlan,
selected_repository: Option<&str>,
) -> (Vec<CodeGraphJob>, Vec<String>) {
let mut anchors_by_repository = BTreeMap::<RepoId, Vec<SymbolAnchor>>::new();
for batch in &focused.source_batches {
let source_path = portable_path(&batch.source.path.display);
let anchors = anchors_by_repository
.entry(batch.source.repo_id.clone())
.or_default();
anchors.extend(batch.outputs.iter().filter_map(|observation| {
if observation.role != SourceRole::Provider {
return None;
}
Some(SymbolAnchor {
symbol: observation.symbol_name.clone()?,
source_path: source_path.clone(),
start_line: usize::try_from(observation.lines.start).ok()?,
})
}));
}
let mut changed_files_by_repository = BTreeMap::<RepoId, Vec<String>>::new();
for change in plan
.changes
.iter()
.filter(|change| change.kind != code_system_graph_model::ArtifactChangeKind::Unchanged)
{
changed_files_by_repository
.entry(change.repo_id.clone())
.or_default()
.push(portable_path(&change.path.display));
}
let mut jobs = Vec::new();
let mut setup_degradations = Vec::new();
for repository in &context.registry.record.repositories {
if selected_repository.is_some_and(|selected| selected != repository.alias) {
continue;
}
let Some(project_path) = context.registry.checkout_path(&repository.alias) else {
setup_degradations.push(format!(
"CodeGraph skipped `{}` because its checkout is unavailable",
repository.alias
));
continue;
};
let mut anchors = anchors_by_repository
.remove(&repository.id)
.unwrap_or_default();
anchors.sort_by(|left, right| {
(&left.source_path, left.start_line, &left.symbol).cmp(&(
&right.source_path,
right.start_line,
&right.symbol,
))
});
anchors.dedup();
if anchors.len() > 50 {
anchors.truncate(50);
setup_degradations.push(format!(
"CodeGraph symbol corroboration for `{}` was limited to 50 anchors",
repository.alias
));
}
let mut changed_files = changed_files_by_repository
.remove(&repository.id)
.unwrap_or_default();
changed_files.sort();
changed_files.dedup();
if changed_files.len() > 1_024 {
changed_files.truncate(1_024);
setup_degradations.push(format!(
"CodeGraph affected-test corroboration for `{}` was limited to 1,024 changed files",
repository.alias
));
}
if !anchors.is_empty() || !changed_files.is_empty() {
jobs.push((
repository.id.clone(),
project_path.to_path_buf(),
anchors,
changed_files,
));
}
}
(jobs, setup_degradations)
}
fn apply_codegraph_corroboration(graph: &mut GraphAssembly, reports: &[RepositoryCorroboration]) {
for item in reports {
for outcome in &item.report.symbols {
let SymbolCorroboration::Confirmed {
symbol,
source_path,
local_id,
} = outcome
else {
continue;
};
let implementation_ids = ["javascript", "typescript", "rust", "python", "go", "java"]
.map(|language| {
NodeId::new(stable_id(
"node",
&format!(
"symbol:{}:{language}:{source_path}:{symbol}",
item.repo_id.as_str()
),
))
})
.into_iter()
.filter(|candidate| {
graph
.nodes
.iter()
.any(|node| node.id == *candidate && node.kind == NodeKind::SymbolRef)
})
.collect::<BTreeSet<_>>();
if implementation_ids.is_empty() {
continue;
}
let source_evidence = graph.evidence.iter().find(|evidence| {
evidence.repo_id.as_ref() == Some(&item.repo_id)
&& evidence.file_path.as_deref() == Some(source_path)
&& evidence.extractor.starts_with("code-system-graph.source.")
});
let (start_line, end_line, content_hash) =
source_evidence.map_or((None, None, None), |evidence| {
(
evidence.start_line,
evidence.end_line,
evidence.content_hash.clone(),
)
});
let evidence_key = format!(
"codegraph:{}:{source_path}:{symbol}:{}",
item.repo_id.as_str(),
local_id.as_deref().unwrap_or("anonymous")
);
let evidence = Evidence {
id: EvidenceId::new(stable_id("evidence", &evidence_key)),
repo_id: Some(item.repo_id.clone()),
file_path: Some(source_path.clone()),
start_line,
end_line,
extractor: "code-system-graph.codegraph.corroboration".to_owned(),
extractor_version: "1.0.0".to_owned(),
provenance: Provenance::CodeGraph,
confidence: 1.0,
observed_at_commit: None,
content_hash,
note: Some("exact provider symbol name, path, and line match".to_owned()),
};
for edge in graph.edges.iter_mut().filter(|edge| {
edge.kind == code_system_graph_model::EdgeKind::ImplementedBy
&& implementation_ids.contains(&edge.target)
}) {
if !edge.evidence.contains(&evidence.id) {
edge.evidence.push(evidence.id.clone());
edge.evidence.sort();
}
}
graph.evidence.push(evidence);
}
}
graph.evidence.sort_by(|left, right| left.id.cmp(&right.id));
graph.evidence.dedup_by(|left, right| left.id == right.id);
}
fn persisted_manual_link_records(
snapshot_id: &str,
decisions: &[LinkDecision],
) -> Result<Vec<ManualLinkRecord>, ApplicationError> {
decisions
.iter()
.map(|decision| {
let disposition = match decision.status {
LinkStatus::Confirmed => ManualLinkDisposition::Active,
LinkStatus::Suppressed => ManualLinkDisposition::Suppression,
LinkStatus::Ambiguous | LinkStatus::Rejected => {
return Err(ApplicationError::Initialization(
"unapplied manual-link decision reached snapshot publication".to_owned(),
));
}
};
let kind = serde_json::to_value(decision.relation)
.ok()
.and_then(|value| value.as_str().map(str::to_owned))
.ok_or_else(|| {
ApplicationError::Initialization(
"manual-link relation could not be encoded".to_owned(),
)
})?;
let reason = decision.reasons.first().cloned().ok_or_else(|| {
ApplicationError::Initialization(
"manual-link decision omitted its required rationale".to_owned(),
)
})?;
let identity = format!(
"{}:{kind}:{}:{}:{reason}",
decision.source.as_str(),
decision.target.as_str(),
match disposition {
ManualLinkDisposition::Active => "active",
ManualLinkDisposition::Suppression => "suppression",
}
);
Ok(ManualLinkRecord {
id: stable_id("manual-link", &identity),
snapshot_id: snapshot_id.to_owned(),
source_node_id: decision.source.clone(),
target_node_id: decision.target.clone(),
kind,
disposition,
reason,
decision: decision.clone(),
config_version: 1,
})
})
.collect()
}
fn provider_capability_record(
workspace: &str,
repo_id: &RepoId,
capability: &ProviderCapability,
) -> ProviderCapabilityRecord {
let mut capabilities = Vec::new();
if let Some(status) = serialized_enum_name(capability.status) {
capabilities.push(format!("status:{status}"));
}
capabilities.extend(
capability
.tools
.iter()
.filter(|tool| tool.len() <= 240)
.map(|tool| format!("tool:{tool}")),
);
for operation in &capability.operations {
if let (Some(operation_name), Some(transport)) = (
serialized_enum_name(operation.operation),
serialized_enum_name(operation.transport),
) {
capabilities.push(format!("operation:{operation_name}:{transport}"));
}
}
capabilities.sort();
capabilities.dedup();
ProviderCapabilityRecord {
workspace_name: workspace.to_owned(),
repo_id: repo_id.clone(),
provider: capability.provider.clone(),
provider_version: capability
.version
.clone()
.unwrap_or_else(|| "unknown".to_owned()),
capabilities,
observed_at_unix_ms: current_unix_millis(),
}
}
fn community_topology_unchanged(
previous_nodes: &[Node],
previous_edges: &[Edge],
current_nodes: &[Node],
current_edges: &[Edge],
) -> bool {
previous_nodes == current_nodes
&& previous_edges.len() == current_edges.len()
&& previous_edges
.iter()
.zip(current_edges)
.all(|(previous, current)| {
previous.id == current.id
&& previous.source == current.source
&& previous.target == current.target
&& previous.kind == current.kind
&& previous.confidence.to_bits() == current.confidence.to_bits()
&& previous.status == current.status
})
}
fn serialized_enum_name<T: Serialize>(value: T) -> Option<String> {
serde_json::to_value(value)
.ok()
.and_then(|value| value.as_str().map(str::to_owned))
}
fn extract_boundaries(context: &WorkspaceContext) -> Result<Vec<HttpBoundary>, ApplicationError> {
let repository_records = context
.registry
.record
.repositories
.iter()
.map(|repository| (repository.alias.as_str(), repository))
.collect::<BTreeMap<_, _>>();
let mut boundaries = Vec::new();
for alias in context.manifest.repos.keys() {
let registered = repository_records
.get(alias.as_str())
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let effective = context
.repository_configs
.get(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
boundaries.extend(
effective
.http_consumers
.iter()
.map(|consumer| HttpBoundary::consumer(registered.id.clone(), consumer)),
);
let repository_path = context
.registry
.checkout_path(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
for openapi in &effective.openapi {
let openapi_path = repository_path.join(openapi);
let mut tracker = ExtractionTracker::new(
openapi,
"code-system-graph.http.openapi",
&context.extraction_budgets,
);
let (openapi_source, _) = read_source_file(&openapi_path, &mut tracker)?;
boundaries.extend(extract_openapi_with_tracker(
®istered.id,
openapi,
&openapi_source,
&mut tracker,
)?);
}
}
Ok(boundaries)
}
fn extract_declared_tests(
context: &WorkspaceContext,
fingerprints: &[ArtifactFingerprint],
) -> Result<Vec<DeclaredTestCase>, ApplicationError> {
let repository_records = context
.registry
.record
.repositories
.iter()
.map(|repository| (repository.alias.as_str(), repository))
.collect::<BTreeMap<_, _>>();
let mut tests = Vec::new();
for alias in context.manifest.repos.keys() {
let repository = repository_records
.get(alias.as_str())
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let effective = context
.repository_configs
.get(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
tests.extend(effective.integration_tests.iter().map(|test| {
let content_hash = artifact_content_hash(
fingerprints,
&repository.id,
"code-system-graph.tests.declared",
&test.path,
);
declared_test_case(repository.id.clone(), test, content_hash)
}));
}
tests.sort_by(|left, right| left.node.id.cmp(&right.node.id));
Ok(tests)
}
fn extract_declared_implementations(
context: &WorkspaceContext,
fingerprints: &[ArtifactFingerprint],
) -> Result<Vec<DeclaredImplementation>, ApplicationError> {
let repository_records = context
.registry
.record
.repositories
.iter()
.map(|repository| (repository.alias.as_str(), repository))
.collect::<BTreeMap<_, _>>();
let mut implementations = Vec::new();
for alias in context.manifest.repos.keys() {
let repository = repository_records
.get(alias.as_str())
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let effective = context
.repository_configs
.get(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
implementations.extend(effective.implementations.iter().map(|implementation| {
let content_hash = artifact_content_hash(
fingerprints,
&repository.id,
"code-system-graph.implementations.declared",
&implementation.path,
);
declared_implementation(repository.id.clone(), implementation, content_hash)
}));
}
implementations.sort_by(|left, right| left.node.id.cmp(&right.node.id));
Ok(implementations)
}
fn artifact_content_hash(
fingerprints: &[ArtifactFingerprint],
repo_id: &code_system_graph_model::RepoId,
extractor: &str,
path: &str,
) -> Option<String> {
fingerprints
.iter()
.find(|fingerprint| {
&fingerprint.repo_id == repo_id
&& fingerprint.extractor == extractor
&& portable_path(&fingerprint.path.display) == portable_path(path)
})
.map(|fingerprint| fingerprint.content_hash.clone())
}
fn portable_path(path: &str) -> String {
path.replace('\\', "/")
}
fn cargo_crate_root(checkout: &Path, source_path: &Path) -> String {
source_path
.parent()
.into_iter()
.flat_map(Path::ancestors)
.take_while(|candidate| candidate.starts_with(checkout))
.find(|candidate| candidate.join("Cargo.toml").is_file())
.and_then(|candidate| candidate.strip_prefix(checkout).ok())
.map(|relative| portable_path(&relative.to_string_lossy()))
.unwrap_or_default()
}
fn discover_artifact_fingerprints(
context: &WorkspaceContext,
) -> Result<Vec<ArtifactFingerprint>, ApplicationError> {
let repository_records = context
.registry
.record
.repositories
.iter()
.map(|repository| (repository.alias.as_str(), repository))
.collect::<BTreeMap<_, _>>();
let mut fingerprints = BTreeMap::new();
for alias in context.manifest.repos.keys() {
let repository = repository_records
.get(alias.as_str())
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let checkout_path = context
.registry
.checkout_path(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
let effective = context
.repository_configs
.get(alias)
.ok_or_else(|| ApplicationError::RegistryAliasMissing(alias.clone()))?;
for openapi in &effective.openapi {
let fingerprint = fingerprint_artifact(
repository,
checkout_path,
Path::new(openapi),
"code-system-graph.http.openapi",
&context.extraction_budgets,
)?;
fingerprints.insert(artifact_key(&fingerprint), fingerprint);
}
for consumer in &effective.http_consumers {
let fingerprint = fingerprint_artifact(
repository,
checkout_path,
Path::new(&consumer.source),
"code-system-graph.http.declared",
&context.extraction_budgets,
)?;
fingerprints.insert(artifact_key(&fingerprint), fingerprint);
}
for test in &effective.integration_tests {
let fingerprint = fingerprint_artifact(
repository,
checkout_path,
Path::new(&test.path),
"code-system-graph.tests.declared",
&context.extraction_budgets,
)?;
fingerprints.insert(artifact_key(&fingerprint), fingerprint);
}
for implementation in &effective.implementations {
let fingerprint = fingerprint_artifact(
repository,
checkout_path,
Path::new(&implementation.path),
"code-system-graph.implementations.declared",
&context.extraction_budgets,
)?;
fingerprints.insert(artifact_key(&fingerprint), fingerprint);
}
for (relative_path, extractor) in
discover_focused_artifacts(checkout_path, &effective.ignore_policy)?
{
let fingerprint = fingerprint_artifact(
repository,
checkout_path,
&relative_path,
extractor,
&context.extraction_budgets,
)?;
fingerprints.insert(artifact_key(&fingerprint), fingerprint);
}
}
Ok(fingerprints.into_values().collect())
}
fn discover_focused_artifacts(
checkout_path: &Path,
ignore_policy: &IgnorePolicy,
) -> Result<Vec<(PathBuf, &'static str)>, ApplicationError> {
let mut pending = vec![checkout_path.to_path_buf()];
let canonical_checkout =
fs::canonicalize(checkout_path).map_err(|source| ApplicationError::ReadFile {
path: checkout_path.to_path_buf(),
source,
})?;
let mut visited = BTreeSet::new();
let mut discovered = Vec::new();
while let Some(directory) = pending.pop() {
let canonical_directory =
fs::canonicalize(&directory).map_err(|source| ApplicationError::ReadFile {
path: directory.clone(),
source,
})?;
if !canonical_directory.starts_with(&canonical_checkout) {
return Err(ApplicationError::ArtifactOutsideCheckout {
path: canonical_directory,
checkout: canonical_checkout,
});
}
if !visited.insert(canonical_directory) {
continue;
}
worker::report_progress(code_system_graph_core::JobPhase::Discovery, 1);
let entries = fs::read_dir(&directory).map_err(|source| ApplicationError::ReadFile {
path: directory.clone(),
source,
})?;
for entry in entries {
let entry = entry.map_err(|source| ApplicationError::ReadFile {
path: directory.clone(),
source,
})?;
let file_type = entry
.file_type()
.map_err(|source| ApplicationError::ReadFile {
path: entry.path(),
source,
})?;
if file_type.is_symlink() {
continue;
}
let path = entry.path();
let relative = path.strip_prefix(checkout_path).map_err(|_| {
ApplicationError::ArtifactOutsideCheckout {
path: path.clone(),
checkout: checkout_path.to_path_buf(),
}
})?;
if file_type.is_dir() {
if !ignore_policy.excludes(relative, true) {
pending.push(path);
}
continue;
}
if !file_type.is_file() || ignore_policy.excludes(relative, false) {
continue;
}
for extractor in focused_extractors_for_path(&path) {
discovered.push((relative.to_path_buf(), extractor));
}
}
}
discovered.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(right.1)));
Ok(discovered)
}
fn focused_extractors_for_path(path: &Path) -> Vec<&'static str> {
let Some(name) = path.file_name().and_then(std::ffi::OsStr::to_str) else {
return Vec::new();
};
let mut extractors = Vec::new();
let is_generated_client_metadata = name == "openapitools.json"
|| (matches!(name, "FILES" | "VERSION")
&& path
.parent()
.and_then(Path::file_name)
.is_some_and(|parent| parent == ".openapi-generator"));
if is_generated_client_metadata {
extractors.push("code-system-graph.http.generated-client");
}
let extension = path
.extension()
.and_then(std::ffi::OsStr::to_str)
.map(str::to_ascii_lowercase);
let source_extractor = match extension.as_deref() {
Some("js" | "jsx") => Some("code-system-graph.source.javascript"),
Some("ts" | "tsx") => Some("code-system-graph.source.typescript"),
Some("rs") => Some("code-system-graph.source.rust"),
Some("py") => Some("code-system-graph.source.python"),
Some("go") => Some("code-system-graph.source.go"),
Some("java") => Some("code-system-graph.source.java"),
_ => None,
};
if let Some(extractor) = source_extractor {
extractors.extend([
extractor,
"code-system-graph.events.source",
"code-system-graph.graphql.source",
"code-system-graph.protobuf.generated",
"code-system-graph.data.source",
]);
}
if matches!(extension.as_deref(), Some("graphql" | "gql")) {
extractors.push("code-system-graph.graphql.document");
}
if extension.as_deref() == Some("proto") {
extractors.push("code-system-graph.protobuf");
}
let lower_name = name.to_ascii_lowercase();
if matches!(
lower_name.as_str(),
"asyncapi.yaml" | "asyncapi.yml" | "asyncapi.json"
) {
extractors.push("code-system-graph.events.asyncapi");
}
if matches!(
lower_name.as_str(),
"persisted-queries.json"
| "persisted_queries.json"
| "apollo-manifest.json"
| "operation-manifest.json"
) {
extractors.push("code-system-graph.graphql.persisted");
}
extractors.extend(document_extractors_for_path(
path,
name,
extension.as_deref(),
));
let is_package_artifact = matches!(
name,
"package.json"
| "package-lock.json"
| "pnpm-lock.yaml"
| "yarn.lock"
| "pyproject.toml"
| "poetry.lock"
| "Cargo.toml"
| "Cargo.lock"
| "go.mod"
| "go.work"
| "pom.xml"
| "build.gradle"
| "build.gradle.kts"
| "packages.config"
) || extension.as_deref() == Some("csproj")
|| (name.starts_with("requirements")
&& Path::new(name)
.extension()
.is_some_and(|extension| extension.eq_ignore_ascii_case("txt")));
if is_package_artifact {
extractors.push("code-system-graph.packages");
}
extractors.sort_unstable();
extractors.dedup();
extractors
}
fn document_extractors_for_path(
path: &Path,
name: &str,
extension: Option<&str>,
) -> Vec<&'static str> {
let mut extractors = Vec::new();
let lower_name = name.to_ascii_lowercase();
let lower_components = path
.components()
.filter_map(|component| component.as_os_str().to_str())
.map(str::to_ascii_lowercase)
.collect::<Vec<_>>();
let in_database_migrations = lower_components
.iter()
.any(|component| matches!(component.as_str(), "alembic" | "migrations"));
if matches!(extension, Some("sql" | "prisma"))
|| lower_name == "sqlx.toml"
|| lower_name == "schema.rs"
|| (extension == Some("py")
&& (in_database_migrations || matches!(lower_name.as_str(), "model.py" | "models.py")))
{
extractors.push("code-system-graph.data.artifact");
}
let is_compose = matches!(
lower_name.as_str(),
"compose.yaml"
| "compose.yml"
| "compose.json"
| "docker-compose.yaml"
| "docker-compose.yml"
| "docker-compose.json"
);
if is_compose {
extractors.push("code-system-graph.infrastructure.compose");
}
if extension == Some("tf") {
extractors.push("code-system-graph.infrastructure.terraform");
}
let in_helm_templates = lower_components
.iter()
.any(|component| component == "templates");
let is_helm = in_helm_templates
|| matches!(
lower_name.as_str(),
"chart.yaml" | "chart.yml" | "values.yaml" | "values.yml"
);
if is_helm {
extractors.push("code-system-graph.infrastructure.helm");
} else if matches!(extension, Some("yaml" | "yml" | "json"))
&& !is_compose
&& (lower_components
.iter()
.any(|component| matches!(component.as_str(), "k8s" | "kubernetes" | "manifests"))
|| [
"deployment",
"service",
"ingress",
"statefulset",
"daemonset",
"cronjob",
"job",
"pod",
]
.iter()
.any(|kind| lower_name.starts_with(kind)))
{
extractors.push("code-system-graph.infrastructure.kubernetes");
}
if extension == Some("md") || lower_name == "markdown" {
extractors.push("code-system-graph.documents.markdown");
}
if lower_name == "codeowners" {
extractors.push("code-system-graph.documents.codeowners");
}
if matches!(
lower_name.as_str(),
"catalog.yaml"
| "catalog.yml"
| "catalog.json"
| "catalog-info.yaml"
| "catalog-info.yml"
| "service-catalog.yaml"
| "service-catalog.yml"
| "service-catalog.json"
) {
extractors.push("code-system-graph.documents.catalog");
}
let is_named_config = [
"config.",
"settings.",
"application.",
"values.",
"secrets.",
]
.iter()
.any(|prefix| lower_name.starts_with(prefix));
if lower_name == ".env"
|| lower_name.starts_with(".env.")
|| (is_named_config && matches!(extension, Some("yaml" | "yml" | "json" | "toml")))
{
extractors.push("code-system-graph.config.safe");
}
extractors
}
fn fingerprint_artifact(
repository: &RepositoryRecord,
checkout_path: &Path,
relative_path: &Path,
extractor: &str,
budgets: &ExtractionBudgets,
) -> Result<ArtifactFingerprint, ApplicationError> {
let configured_path = checkout_path.join(relative_path);
let canonical_path =
std::fs::canonicalize(&configured_path).map_err(|source| ApplicationError::ReadFile {
path: configured_path.clone(),
source,
})?;
if !canonical_path.starts_with(checkout_path) {
return Err(ApplicationError::ArtifactOutsideCheckout {
path: configured_path,
checkout: checkout_path.to_path_buf(),
});
}
let metadata =
std::fs::metadata(&canonical_path).map_err(|source| ApplicationError::ReadFile {
path: canonical_path.clone(),
source,
})?;
let relative = canonical_path.strip_prefix(checkout_path).map_err(|_| {
ApplicationError::ArtifactOutsideCheckout {
path: canonical_path.clone(),
checkout: checkout_path.to_path_buf(),
}
})?;
let path = encode_native_path(relative);
code_system_graph_model::validate_safe_path_display(&path.display)
.map_err(|_| ApplicationError::UnsafeArtifactPath)?;
let mut tracker = ExtractionTracker::new(&path.display, extractor, budgets);
let content = read_bounded_bytes(&canonical_path, &mut tracker)?;
let fingerprint = ArtifactFingerprint {
repo_id: repository.id.clone(),
checkout_id: repository.checkout_id.clone(),
path,
extractor: extractor.to_owned(),
content_hash: stable_id_bytes("artifact-content", &content),
size_bytes: metadata.len(),
};
worker::report_progress(code_system_graph_core::JobPhase::Fingerprinting, 1);
Ok(fingerprint)
}
fn artifact_key(
fingerprint: &ArtifactFingerprint,
) -> (CheckoutId, code_system_graph_model::NativePath, String) {
(
fingerprint.checkout_id.clone(),
fingerprint.path.clone(),
fingerprint.extractor.clone(),
)
}
fn extractor_runs(
snapshot_id: &str,
fingerprints: &[ArtifactFingerprint],
plan: &IncrementalPlan,
) -> Vec<ExtractorRun> {
let actions = plan
.changes
.iter()
.map(|change| {
(
ArtifactKey {
repo_id: change.repo_id.clone(),
checkout_id: change.checkout_id.clone(),
path: change.path.clone(),
extractor: change.extractor.clone(),
},
change.kind,
)
})
.collect::<BTreeMap<_, _>>();
let mut groups = BTreeMap::<(String, String, String), Vec<&ArtifactFingerprint>>::new();
for fingerprint in fingerprints {
groups
.entry((
fingerprint.repo_id.as_str().to_owned(),
fingerprint.checkout_id.as_str().to_owned(),
fingerprint.extractor.clone(),
))
.or_default()
.push(fingerprint);
}
groups
.into_iter()
.map(|((repo_id, checkout_id, extractor), inputs)| {
let changed = inputs
.iter()
.filter(|fingerprint| {
actions.get(&ArtifactKey::from(**fingerprint))
!= Some(&code_system_graph_model::ArtifactChangeKind::Unchanged)
})
.count();
let discovered_files = u64::try_from(inputs.len()).unwrap_or(u64::MAX);
let parsed_files = u64::try_from(changed).unwrap_or(u64::MAX);
let skipped_files = discovered_files.saturating_sub(parsed_files);
let key = format!("{snapshot_id}:{repo_id}:{checkout_id}:{extractor}");
ExtractorRun {
id: stable_id("extractor-run", &key),
snapshot_id: snapshot_id.to_owned(),
repo_id: RepoId::new(repo_id),
checkout_id: CheckoutId::new(checkout_id),
extractor_version: if focused_extractor(&extractor) {
EXTRACTION_CONTRACT_VERSION.to_owned()
} else {
env!("CARGO_PKG_VERSION").to_owned()
},
extractor,
status: if changed == 0 {
ExtractorRunStatus::SkippedUnchanged
} else {
ExtractorRunStatus::Success
},
discovered_files,
parsed_files,
skipped_files,
elapsed_ms: 0,
}
})
.collect()
}
fn read_file(path: &Path) -> Result<String, ApplicationError> {
std::fs::read_to_string(path).map_err(|source| ApplicationError::ReadFile {
path: path.to_path_buf(),
source,
})
}
fn read_source_file(
path: &Path,
tracker: &mut ExtractionTracker,
) -> Result<(String, bool), ApplicationError> {
let bytes = read_bounded_bytes(path, tracker)?;
match String::from_utf8(bytes) {
Ok(source) => Ok((source, false)),
Err(error) => Ok((String::from_utf8_lossy(error.as_bytes()).into_owned(), true)),
}
}
fn read_bounded_bytes(
path: &Path,
tracker: &mut ExtractionTracker,
) -> Result<Vec<u8>, ApplicationError> {
let metadata = std::fs::metadata(path).map_err(|source| ApplicationError::ReadFile {
path: path.to_path_buf(),
source,
})?;
tracker.check_input_bytes(metadata.len())?;
let file = std::fs::File::open(path).map_err(|source| ApplicationError::ReadFile {
path: path.to_path_buf(),
source,
})?;
let maximum = tracker.budgets().max_input_bytes_per_artifact;
let mut bytes = Vec::new();
file.take(maximum.saturating_add(1))
.read_to_end(&mut bytes)
.map_err(|source| ApplicationError::ReadFile {
path: path.to_path_buf(),
source,
})?;
tracker.check_input_bytes(u64::try_from(bytes.len()).unwrap_or(u64::MAX))?;
Ok(bytes)
}
#[cfg(test)]
mod budget_regression_tests {
use super::*;
#[test]
fn direct_scan_should_preserve_focused_source_value_limit() {
let temporary = tempfile::tempdir().expect("temporary directory");
let repository = temporary.path().join("api");
fs::create_dir_all(repository.join("src")).expect("source directory");
fs::write(
repository.join("src/routes.rs"),
"fn focused_source_boundary() {}",
)
.expect("source fixture");
let config = temporary.path().join("code-system-graph.yaml");
let database = temporary.path().join("code-system-graph.db");
fs::write(
&config,
"version: 1\nname: source-budget-unit\nextractionBudgets:\n maxIdentifierBytesPerValue: 3\nrepos:\n api:\n path: api\n",
)
.expect("manifest fixture");
let expected_artifact = encode_native_path(&Path::new("src").join("routes.rs")).display;
let result = scan_workspace_direct(&config, &database, &ScanOverrides::default());
assert!(
matches!(
&result,
Err(ApplicationError::ExtractionLimit(error))
if error.resource == code_system_graph_core::ExtractionResource::IdentifierBytesPerValue
&& error.artifact == expected_artifact
&& error.extractor == "code-system-graph.source.rust"
),
"unexpected direct source budget result: {result:?}"
);
}
#[test]
fn direct_scan_should_charge_observations_independently_of_syntax_candidates() {
let temporary = tempfile::tempdir().expect("temporary directory");
let repository = temporary.path().join("tests");
fs::create_dir_all(repository.join("tests")).expect("source directory");
fs::write(
repository.join("tests/test_api.py"),
"import requests\ndef test_create_order():\n requests.post(\"https://api.test/v1/orders\")\n",
)
.expect("source fixture");
let config = temporary.path().join("code-system-graph.yaml");
let database = temporary.path().join("code-system-graph.db");
let manifest = |maximum| {
format!(
"version: 1\nname: source-observation-unit\nextractionBudgets:\n maxObservationsPerArtifact: {maximum}\nrepos:\n tests:\n path: tests\n"
)
};
fs::write(&config, manifest(1)).expect("limited manifest");
let rejected = scan_workspace_direct(&config, &database, &ScanOverrides::default());
assert!(matches!(
rejected,
Err(ApplicationError::ExtractionLimit(error))
if error.resource == code_system_graph_core::ExtractionResource::Observations
&& error.observed == 2
&& error.maximum == 1
));
fs::write(&config, manifest(2)).expect("exact manifest");
let accepted = scan_workspace_direct(&config, &database, &ScanOverrides::default());
assert!(
accepted.is_ok(),
"exact observation budget failed: {accepted:?}"
);
}
}
#[cfg(test)]
mod codex_review_regression_tests {
use code_system_graph_model::validate_safe_path_display;
use super::ApplicationError;
#[test]
fn unsafe_artifact_path_error_must_not_echo_rejected_display() {
const BIDI_PAYLOAD: &str = "invoice\u{202e}pay.pdf";
assert!(validate_safe_path_display(BIDI_PAYLOAD).is_err());
let rendered = ApplicationError::UnsafeArtifactPath.to_string();
assert!(!rendered.contains(BIDI_PAYLOAD));
assert!(!rendered.contains('\u{202e}'));
assert_eq!(
rendered,
"artifact path contains unsafe control or bidirectional characters"
);
}
}