mirror of
https://github.com/Kaelio/ktx.git
synced 2026-07-01 08:59:39 +02:00
* docs: add spider2-specs handoff directory for benchmark-driven feature specs
* feat(cli): connection-scoped wiki pages
Add an optional `connections` frontmatter field so database-specific wiki
knowledge can be scoped to a connection without polluting searches about other
databases, while page keys stay a flat, globally-unique namespace.
- connections: single string or list; absent/empty ⇒ unscoped (applies to all)
- wiki_search (MCP) and `ktx wiki --connection` return unscoped ∪ matching
pages, filtered at the disk-load seam so all three search lanes draw their
candidate pool from the already-scoped set (not a post-filter)
- wiki_write accepts connections with REPLACE semantics and rejects a
connection-scoped write whose key collides with a disjoint-connection page
(data-loss guard; hard error, no silent clobber)
- explicit connection-id args (wiki_search, memory_ingest, ktx wiki) are
validated against ktx.yaml via a shared assertConfiguredConnectionId, which
also closes the prior gap where memory_ingest's connectionId was unvalidated;
persisted ids absent from config warn (not fail) in `ktx status`
- prompt guidance in the wiki_capture skill and external-ingest prompt; the
session connectionId is surfaced to the memory agent and ingest work units
Implements spider2-specs/specs/01-connection-scoped-wiki.md; intake draft moved
to spider2-specs/done/.
* docs(spider2-specs): add specs/ refinement stage and composite-key join spec
Describe the todo/ → specs/ → done/ pipeline in the README (refined specs are
the durable artifact; intake drafts move to done/ on ship) and add a
MEDIUM-priority spec for multi-column composite-key join detection found during
the first sqlite smoke test.
* feat(cli): add --verbatim ingest mode for authoritative documents
Store each --text/--file document body unchanged as a GLOBAL wiki page
instead of routing it through the memory agent, which may rewrite,
condense, or re-title it. The LLM derives only metadata (summary, tags,
sl_refs) and only for frontmatter fields the document does not already
set; the stored body is written by code and never edited.
- Deterministic page key: files derive it from the filename, inline
text from its leading Markdown heading (headless inline text is
rejected — pass it as --file instead).
- Idempotent: re-running the same body is a no-op; a different body at
the same key fails loudly rather than overwriting.
- Works with llm.provider.backend: none, deriving a degraded summary
from the heading or first sentence.
- Existing frontmatter (including unmodeled fields like effective_date)
passes through untouched; --connection-id scopes the page.
* feat(cli): SQL-authoring craft and per-dialect notes tool for the analytics skill
Spec 07: add a dialect-agnostic <sql_craft> block to the ktx-analytics skill (schema discovery, composition, window-function correctness, numeric precision, answer completeness) with one worked window-then-filter example. Workflow steps gain pointers into it; existing guidance is unchanged.
Spec 08: add a read-only sql_dialect_notes MCP tool returning a connection's engine SQL conventions (FQTN form, identifier quoting/case, date/time, top-N idiom, JSON access), resolved through the existing sqlAnalysisDialectForDriver path. Notes are per-dialect markdown files under context/sql-analysis/dialects, served by the tool and copied to dist (package-internal, never installed). Non-SQL connections return a clear KtxExpectedError. The flat skill gains a one-line pointer to the tool.
Both spider2-specs intake drafts move to done/ with implementation notes.
* feat(cli): tolerate objects that fail introspection during scan
Isolate per-object introspection failures so one broken or inaccessible object no longer zeroes out a connection's whole semantic layer: the sqlite and bigquery connectors introspect each object defensively (tryIntrospectObject), the live-database adapter records a scan outcome and fetch report, and enabled_tables accepts catalog.db.name, db.name, or bare names with a clear no-match error. Includes matching ktx-daemon introspection changes, docs, and tests.
* docs(spider2-specs): add 06-scan-tolerate-broken-objects spec
* feat(cli): generalize analytics fan-out rule to multi-hop join chains
The ktx-analytics skill's fan-out rule only reliably caught single-hop
inflation; agents still silently fanned out on multi-hop chains where the
offending one-to-many join sits several hops below the SUM/COUNT and is easy
to miss.
Rewrite the Composition rule so the danger reads as cumulative across the whole
chain (pre-aggregate per measure-owning table), add an affirmative
grain-verification habit (default: pre-aggregate to grain; escape hatch:
COUNT(DISTINCT key) for pure counts only; SUM/AVG of a fanned-out measure must
pre-aggregate), and add one generic wrong-vs-right worked example. Content-only
and dialect-agnostic; no new tool, flag, or config.
Implements spider2-specs/specs/09 and annotates spec 07's one-example
constraint as superseded.
* feat(cli): add panel-completeness, time-series window, and text-encoded numeric SQL craft
Extend the analytics skill's <sql_craft> with three correctness habits and
route the dialect-specific halves through sql_dialect_notes:
- Panel completeness (spec 10): full-domain spine -> LEFT JOIN -> COALESCE for
"each/every/all/per" questions, defaulted by measure additivity.
- Time-series windows (spec 11): explicit cumulative frames, calendar-range
rolling windows with minimum-periods guards, and period-over-period via LAG.
- Text-encoded numerics (spec 12): sample distinct values, strip/scale/cast in
one early CTE, and confirm coverage with a failure-detecting cast.
Add per-dialect Series, Rolling window, and Safe cast notes to all seven
dialect files so the skill stays dialect-agnostic while the engine-specific
syntax lives in sql_dialect_notes. Tests updated and passing (19).
* docs(spider2-specs): add specs 10-12 for analytics SQL-craft additions
Refined specs and completion records for the panel-completeness spine (10),
time-series window recipes (11), and text-encoded numeric parsing (12)
implemented in the preceding commit.
* docs(spider2-specs): add backlog intake drafts 13-14
- 13: canonical authoritative-source measures
- 14: output-completeness final check
* skill(analytics): spec 14 output-completeness + iter1 (active column planning)
Bundles two changes (entangled in SKILL.md; future spider2 iterations land as
separate commits):
- spec 14 (output-completeness): multi-part "answer every requested output" rule
+ a "Final completeness check" in workflow Step 6 and <sql_craft>; analytics
skill-content test updated; intake draft -> done/, refined spec added.
- iter1 experiment: spec 14's passive end-check did not change behavior on the
benchmark's output-completeness failures, so (a) the Plan step now writes the
exact output-column list UP FRONT as a contract the final SELECT must match,
and (b) "expose identity" -> "project BOTH the entity id and its name" (covers
both omission directions). All generic craft.
Driven by the Spider 2.0-Lite failure analysis (incomplete output was the
largest failure bucket); benchmark only as motivation.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* skill(analytics): iter2 — deterministic order in string/array aggregation
GROUP_CONCAT/string_agg/array_agg element order is undefined without an explicit
ORDER BY; also note SQLite's default text sort is binary/case-sensitive (uppercase
before lowercase) vs case-insensitive (COLLATE NOCASE). Generic SQLite craft.
Spider 2.0-Lite motivation: an ordered-ingredient-list question failed only on the
within-string element order (right elements, wrong order); benchmark as motivation only.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* feat(mcp): structured, leveled logging for the MCP server
Add one synchronous pino logger per MCP server process, written through the
io.stderr sink: plain JSON when stderr is not a TTY, colorized pino-pretty
(sync, in-process) when it is. Every tool call logs tool.start with its raw
params BEFORE the handler runs and tool.end after (info / warn past
KTX_MCP_SLOW_TOOL_MS / error), correlated by callId plus sessionId, so a
runaway sql_execution leaves a recoverable start line with its exact SQL and
no matching end. HTTP logs session.open/close and wires the previously-dead
transport.onerror to transport.error; stdio routes its transport error
through the logger. Level via KTX_MCP_LOG_LEVEL (default info). Existing
mcp_request_completed telemetry and registerParsedTool are unchanged; no
worker/async transport and no redaction in v1 (logs are local-only).
Implements spider2-specs/specs/15-mcp-server-structured-logging.md and moves
the intake draft to done/.
* feat(mcp): report uptimeMs in MCP server /health
The /health endpoint now includes uptimeMs (monotonic elapsed time since
the server started), mirroring the Python daemon's uptime_ms telemetry
field.
* feat(cli): bound read-query execution with a per-connection deadline
Enforce one shared query deadline (default 30s, overridable per connection via
query_timeout_ms) on every executeReadOnly path, so an accidentally-expensive
LLM-authored query returns a fast "query exceeded Ns" KtxQueryError instead of
hanging the MCP server.
- New shared contract context/connections/query-deadline.ts
(resolveQueryDeadlineMs, queryDeadlineExceededError); query_timeout_ms added to
the shared warehouse schema; BigQuery's job_timeout_ms removed.
- SQLite runs the read query in a short-lived forked child process and enforces
the deadline with SIGKILL. worker_threads + terminate() was tried first but
cannot interrupt a synchronous better-sqlite3 scan (the native loop never
yields); SIGKILL reclaims the process in ~2ms and keeps the event loop free.
- Remote connectors apply a real server-side statement timeout and re-wrap their
own timeout signal as KtxQueryError: Postgres statement_timeout/57014, MySQL
max_execution_time/3024, Snowflake STATEMENT_TIMEOUT_IN_SECONDS/604, ClickHouse
max_execution_time + aligned request_timeout/159, SQL Server requestTimeout/
ETIMEOUT, BigQuery jobTimeoutMs.
- Relationship validation skips a candidate to review on a deadline timeout
instead of aborting the pass; the deadline surfaces through the existing MCP
pino logger as a matched tool.start/tool.end(error) pair (no new logging code).
Also fixes a pre-existing, unrelated invalid cast in mcp-server-factory.test.ts
that was breaking tsc -p tsconfig.test.json.
* docs(spider2-specs): mark spec 16 (bounded query execution) done
Append Implementation notes to the refined spec (what shipped, where, and the
worker-thread -> child-process+SIGKILL deviation with its evidence) and move the
intake draft from todo/ to done/.
* skill(analytics): iter3 — measure-as-amount, inter-event gap, top-per-metric career
Three generic interpretation rules: a named business measure (sales/revenue/spend)
means its amount not a row count; "inter-event duration/gap" is LAG/LEAD time-between
events not a magnitude column; "highest across several achievements" aggregates per
metric over the whole history. All three demonstrably FIRE (verified on local008/003/152
SQL). local008 flips to correct (mechanism-aligned). 003/152 still fail on a different
axis (source-column / grouping). Generic craft; benchmark only as motivation.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* skill(analytics): spine-for-extreme-selection + aggregate-over-selected-set
Two generic answer-completeness refinements:
- Selecting the extreme group (lowest/highest count over a period/category
domain) must rank over the COMPLETE spine, not only groups with fact rows —
an empty period is a genuine 0 and often the true minimum.
- An aggregate scoped to a per-entity selected set ('avg revenue per actor in
those top-3 films') is computed ACROSS that set, distinct from the per-item
value; project both.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): iter2 — sharpen extreme-selection spine + top-N ranking-measure
- spine-for-extreme: concrete cue that a zero-row period never appears in a
GROUP BY of the facts; generate the full calendar, LEFT JOIN, COALESCE, then rank.
- aggregate-over-selected-set: top-N selection ranks by the named ranking measure
(the item's own revenue), independent of the per-item share that feeds the aggregate.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): iter3 — comparison-between-two-extremes is one wide row
Distinguishes a cross-item comparison ('the difference between the highest and
lowest month' -> single wide row, both extremes side by side + the comparison
column) from 'report a metric for each group' (-> stays long). Generic, question-
derived; targets the wide-vs-long shape gap without affecting per-group long output.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): iter4 — anchor a period bucket to the named lifecycle event
When a record carries multiple lifecycle timestamps (created/placed, approved,
shipped, delivered, completed, settled) and the question counts/measures records
in a named *completed state* by period ("delivered orders by month", "shipped
items per week"), bucket the period by that named event's own timestamp, not the
record-creation timestamp; the state value is the qualifying filter, the matching
timestamp is the time anchor. Wording priority is explicit — purchased/placed/
created/submitted/ordered keep the start-event timestamp — and a non-temporal
state filter (counts by customer/city/seller with no period) introduces no anchor.
Generic analytics craft: counting completed-state records by their creation date
silently answers "records that later reached that state, grouped by when they
started" instead of the question asked. Surfaced via the spider2-autofix loop;
FAIR_PRODUCT (adversary-screened, restatable from question wording + schema/
semantic-layer lifecycle descriptions, no gold dependency).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): iter5 — canonicalize observed URL-path variants before page-level analysis
When a question groups/filters/sequences web pages by a path/url column, sample
its distinct values; if the data itself shows /route and /route/ variants for the
same page context, canonicalize in an early CTE (preserve / as root, strip trailing
slashes from non-root paths, map an observed empty path to / only when the column is
a URL path with blank root-page events) and use the canonical path everywhere above.
Explicitly forbids inventing aliases the data doesn't show: no merging different
route names, no stripping query/fragment/host/scheme, no lowercasing, and no
canonicalization when the question asks for raw URL/path or slash-vs-no-slash diffs.
Generic web-analytics craft: raw request logs routinely store the same user-visible
page with and without a trailing slash, so grouping raw labels silently splits one
page into several. Surfaced via the spider2-autofix loop (Codex runner, round r2);
FAIR_PRODUCT (adversary-screened, restatable from URL-path semantics + page-grain
question wording + solver-observed distinct values, no gold dependency). The rule
fired mechanism-aligned on both targets; flipped local330 (landing/exit page counts),
local331 residual is a separate sequence-semantics axis beyond canonicalization.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): iter6 — coverage over a selected group is a set-membership aggregate
When a question first selects a group of entities ("the top 5 actors", "these
products") and then asks what count/share/percentage of a DIFFERENT subject domain
relates to *these* selected entities ("what % of customers rented films featuring
these actors"), the subject set is the UNION across the whole group: count DISTINCT
subject ids once across the selected entities and return one collective value at the
subject-domain grain — not one row per selected entity (which double-counts subjects
related to more than one entity and answers a different question). Narrowly guarded:
emit one row per entity only when the wording says "for each / per / by / list" or
asks for each entity's own metric ("top 5 players and their batting averages").
The collective-coverage cousin of the existing per-entity selected-set rule. Generic
analytics craft (per-entity metric vs set-level coverage). Surfaced via the
spider2-autofix loop (Codex runner, round r3); FAIR_PRODUCT (adversary-screened,
restatable from wording alone, no gold dependency). Flipped local195 mechanism-aligned
(union COUNT(DISTINCT customer)/total, one scalar); 0 regression across 5 passing
per-entity top-N guards (local023/024/029/212/221 stayed long).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* skill(analytics): label-only joins must LEFT JOIN — incomplete dims silently drop fact rows
Mirror of the existing fan-out rule for the DROP direction: an inner JOIN to a
dimension table used only to attach a display attribute silently discards every
fact row whose key has no parent when the dimension is incomplete (trimmed
catalogs, late-arriving / SCD-gap rows), shrinking counts/sums and the universe
over which shares/averages/medians are computed. Guidance: LEFT JOIN pure
enrichment; inner-join a dimension only when intended as a filter; key the
aggregate/GROUP BY on the fact column, not the dimension column.
Spider2 autofix round 'joindim': flips complex_oracle local050 (FAIL->PASS,
official scorer) — solver dropped the gratuitous products inner-join and
recovered the exact gold. local060/063 also adopt LEFT JOIN (rule fires) but
remain gold-convention-blocked. Guards local061/067 held.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* docs(spider2-specs): add todo/17 — lifecycle-event metrics (semantic-layer)
Draft intake spec surfaced by the spider2-autofix loop (round r1): the model-layer
form of the shipped iter4 lifecycle-date-anchoring skill rule — infer per-state
lifecycle-event metrics (e.g. delivered_orders with defaultTimeDimension = the
delivery timestamp) during enrichment so the correct time anchor is the default for
any consumer, not only an agent that loaded the skill. Generic; FAIR_PRODUCT.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* fix(connectors): accept leading underscore in connection/identifier ids
The safe-identifier validator regex /^[a-zA-Z0-9][a-zA-Z0-9_-]*$/ allowed an
underscore everywhere except the first character, so a connection id / database
name that legitimately starts with '_' (valid in Snowflake, e.g. _1000_GENOMES)
could never be ingested or queried. Allow a leading underscore across all 16
duplicated validators (connection ids, source ids, page/wiki keys, warehouse-
verification tool schemas). Path-safety is unaffected — '.' and '/' remain
excluded, and assertSafePathToken still blocks traversal.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(analytics): generic geospatial query guidance
Add a Snowflake ST_* dialect note (ST_MAKEPOINT lon-first, ST_DWITHIN/ST_CONTAINS/
ST_WITHIN/ST_INTERSECTS, bbox->polygon via ST_MAKEPOLYGON/ST_MAKELINE) and a
dialect-agnostic 'Spatial predicates' recipe in the analytics skill (resolve the
entity geometry, build an area-of-interest polygon, test with the engine's
containment/proximity/overlap predicate; mind lon/lat argument order). Steers the
solver off hand-rolled lat/lon BETWEEN boxes toward correct, index-assisted
geospatial predicates.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(analytics): parse code/dependency text by language grammar
Add two generic <sql_craft> rules: (1) parse imported/required/loaded packages by
the language or manifest format (Java import keep-package-path allowing underscores/
mixed-case; Python import/from + alias stripping; R library/require; .ipynb parse
JSON cell source before language rules; JSON manifests flatten the dependency object
keys), stripping comments/prose and splitting multi-import lines; (2) on a
de-duplicated table with a documented copy/occurrence count, choose COUNT(*) vs the
weight column from the population the question names, not silently. Steers off one
broad regex that drops valid identifiers and matches prose.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(analytics): source filters/dates/measures from the owning fact grain
Add a <sql_craft> rule for joined fact tables at different grains (parent order
vs child line item): read each predicate, calendar bucket, and measure from the
table whose grain the question names, not whichever is in scope post-join. An
order-grain filter ("orders that are Complete", "the order's creation date")
must come from the parent even though the child carries its own status/created_at;
line price/cost come from the child. Mirror at metric grain: don't combine a
parent-grain count with child rows (num_of_item * SUM(line_price) per line) —
aggregate each measure at its own grain before combining.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(analytics): collapse multi-valued classes to one representative per entity before counting/concentration
When an entity carries a multi-valued classification array (IPC/CPC codes, tags)
and the methodology counts entities-per-class or a concentration/diversity metric
(HHI, originality, share), pick ONE representative per entity first (the array's
main/primary/first flag, else a defined fallback like most-frequent), then
aggregate; and use COUNT(DISTINCT entity) when the denominator is defined as a
count of entities. Unnesting the array otherwise multiplies an entity's weight by
its code count, inflating per-class frequencies and skewing the ranking/score.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(connectors): introspect BigQuery datasets hosted in foreign projects
A dataset_ids/dataset_id entry may now be written `project.dataset` to
introspect a dataset hosted in another project while query jobs still bill to
credentials.project_id. Entries are parsed once at the config boundary into
canonical {project, dataset} pairs; introspection, primary-key discovery,
testConnection, getTableRowCount, and listTables (grouped per project) all
resolve in the dataset's own project, and scanned tables are labeled with that
project so sampling, distinct-value, and read queries resolve. Bare entries are
unchanged.
Implements spider2-specs/specs/18-bigquery-cross-project-datasets.md.
* feat(scan): durable, resumable, bounded relationship detection during enrichment
Move the enrichment persistence boundary to the cost boundary and bound the
open-ended relationship stage (spec 19).
- Checkpoint descriptions + embeddings into the queryable `_schema` manifest
(and the raw enrichment artifacts) before relationship detection runs, via a
new `onCheckpoint` hook + `writeLocalScanEnrichmentCheckpoint`. An interrupted,
budget-truncated, or failed relationship stage now degrades to "no joins",
never "no descriptions".
- Resume the enrichment cache by content identity: re-key the SQLite stage store
on `(connection_id, stage, input_hash)` so a re-run with a fresh runId resumes
finished descriptions/embeddings instead of re-paying for LLM work. The
disposable cache recreates its table if the on-disk key shape differs.
- Make the relationship stage observable and bounded: a sticky wall-clock budget
(`scan.relationships.detectionBudgetMs`, default 600000 ms) + per-unit progress
+ honored `ctx.signal`, threaded through profiling, validation, and composite
detection. On exhaustion/abort it stops scheduling, finalizes, and returns a
partial result instead of throwing or hanging.
- Mark a budget/abort-truncated result partial (diagnostics `partial`/`partialReason`
+ recoverable `relationship_detection_partial` warning). A graceful partial saves
as a completed stage and resumes cheaply; raising the budget changes inputHash
and forces a fresh, fuller run. A process killed mid-stage saves nothing.
Document `detectionBudgetMs` in the ktx.yaml reference. Append implementation
notes to specs/19 and move the intake draft to done/.
Also carries the in-tree per-table enrichment LLM timeout work it builds on
(`description-generation.ts` + the `enrichment_timeout` warning code), which is
intertwined in `local-enrichment.ts`/`types.ts` and cannot be split into a
separately-building commit.
* feat(scan): bound + retry the per-table enrichment LLM call
The batched table-description call had no retry (sampleTable retried 3x, this did
not), so a single transient backend error (e.g. an overloaded/burst rejection when
many tables enrich concurrently) silently nulled a whole table's descriptions —
observed dropping ~70% of a db's tables during a bad window despite ample quota.
- Wrap generateObject in retryAsync (3 attempts + backoff; KTX_ENRICH_LLM_ATTEMPTS).
- Fresh per-attempt timeout (KTX_ENRICH_LLM_TIMEOUT_MS, default 120s) still bounds a
wedged wide table; a timeout is surfaced as KtxAbortedError so it is NOT retried
(one wedge stays one timeout, not 3x).
- Granular per-table progress + start/done/retry/timeout logging.
Composes with spec 19 (its non-goal #1): spec 19 makes completed descriptions durable;
this makes more of them complete.
* feat(scan): survive a hung LLM enrichment backend and resume descriptions
Two compounding failure modes on the per-table description-enrichment path (spec 20):
Enforced per-table timeout for subprocess backends. The runtime declares whether it owns an SDK subprocess (subprocessForkSpec on KtxLlmRuntimePort); codex/claude-code calls run behind a ktx-owned detached child that is tree-killed (SIGKILL of the process group on POSIX, taskkill /T on Windows) on the deadline or ctx.signal, reaping the wedged model grandchild. HTTP backends keep native fetch abort. Default stays 120s, one-wedge-one-timeout.
Incremental, resumable descriptions persistence. generateDescriptions flushes enriched tables per batch to an inputHash-tagged durable record (at a stable, non-syncId path) plus only the changed manifest shards, skips already-enriched tables on resume, and never lets one table's failure discard the stage (a skipped table costs one missing description, not the whole stage's output).
Spec 20 refined + intake draft moved to done/.
* feat(scan): selective enrichment stages (--stages) + per-stage cache keys
Split the single coarse enrichment cache key into per-stage hashes
(descriptions <- snapshot + LLM identity; embeddings <- snapshot + embedding
identity + description digest; relationships <- snapshot + relationship settings
+ LLM identity), so changing one stage's inputs invalidates only that stage and
never throws away the expensive per-table descriptions on an unrelated edit.
Add `ktx ingest --stages <list>` to force-re-run a chosen subset on an
already-ingested connection: a named stage bypasses the completed-stage
short-circuit while the per-table descriptions resume record still skips
already-enriched tables, and unselected stages are left untouched on disk. Feed
embeddings + relationships their description context from the on-disk _schema
when descriptions do not run this invocation, and carry descriptions into the
llmProposals evidence packet (closing a latent gap on the full-run path too).
Surface an enrichment_stage_stale warning when an unselected stage's inputs have
drifted, rather than silently cascading the work.
Implements spider2-specs/specs/21-selective-enrichment-stages.md.
* test(analytics): realign SKILL.md acceptance test with the evolved skill
Three assertions in analytics-skill-content.test.ts drifted from the analytics
SKILL.md as later iterations edited the skill without updating the test:
- the sub-heading was renamed Window functions -> Ordering & aggregation
determinism (iter2), so follow the source name;
- the rule "Expose identity, not just the label" was renamed to "Project BOTH
identity and label" (spec 14), so match the new wording;
- the dialect-FQTN guard false-positived on the Java package example
com.planet_ink.coffee_mud, whose backticks made a 3-segment package path read
as a BigQuery/Snowflake `a.b.c` table reference. Drop the backticks so the
guard stays at full strength without weakening it.
* fix(scan): --stages subset must not delete unselected stages' on-disk artifacts
A --stages subset that omitted descriptions wiped all on-disk ai/db descriptions
from the written _schema. runLocalScan writes the structural manifest shard from
the bare snapshot BEFORE enrichment runs, and the shard merge treats ai/db as
scan-managed and overwrites them with whatever the run emits — none, on a subset
that skips descriptions. Enrichment then read the already-wiped shard via
loadPriorDescriptions and had nothing to restore.
runLocalScanEnrichment now returns the best-available descriptions (fresh-this-run
if descriptions ran, else loaded from the on-disk _schema) instead of [], and
runLocalScan captures the prior descriptions before the structural write and feeds
them to both the structural write and enrichment, so an unselected stage's
artifacts survive. Joins were already preserved for --stages descriptions via the
manual/inferred preservedJoins path.
Tests: a full runLocalScan --stages relationships path test (RED without the fix,
GREEN with it — the earlier unit test missed the structural-pre-write ordering),
plus enrichment-layer contract tests for both directions. Validated live on
northwind: --stages relationships keeps all 110 descriptions + 22 joins (was
wiping to 0); --stages descriptions restores descriptions from the spec-20 resume
record (no LLM calls) while keeping joins.
* feat(dialects): bigquery nested-data (ARRAY/STRUCT/UNNEST), geospatial (GEOGRAPHY), SAFE_DIVIDE
bigquery.md lacked the two sections that define BigQuery analytics (present in snowflake.md):
- Nested & repeated data: UNNEST to flatten arrays of STRUCTs (GA360 hits, GA4 event_params),
dot-notation field access, key-value param scalar-subquery extraction, fan-out/COUNT(DISTINCT) guard.
- Geospatial (GEOGRAPHY): ST_GEOGPOINT (lon-first), containment/proximity/distance/intersection
predicates, areal allocation via ST_AREA(ST_INTERSECTION()).
- SAFE_DIVIDE for zero-denominator-safe rates; sharded-table shard-presence note.
Generic BigQuery craft surfaced by sql_dialect_notes; product-completeness (any BQ analyst benefits).
* spec(ingest): resumable + fault-tolerant source ingest (#22)
Refined spec for two source-ingest durability gaps surfaced by a real
user report on a ~2-day dbt ingest: (1) interrupted runs restart every
work unit from scratch (no cross-run reuse), and (2) the final
integration gate is all-or-nothing — one unfixable artifact discards the
whole run.
Design: automatic content-keyed work-unit resume reusing the scan
durability primitive (specs 19/20), plus a deterministic dangling-edge
prune that replaces the fatal final-gate throw so a single bad model
costs only that model, not the run. Prune operates on the integrated
tree and never poisons the cache, so resume and prune self-heal.
* refactor(scan): route enrichment resume through shared cache
* feat(ingest): replay cached work unit patches
* refactor(ingest): return structured final gate findings
* feat(ingest): prune final gates without LLM repair
* docs(ingest): document final gate pruning
* test(ingest): cover stale work unit cache recompute
* fix(ingest): refresh stale cache recompute metadata
* test(ingest): cover missing-target prune and self-heal
* fix(ingest): defer pruneable final gate findings
* fix(ingest): replay pruned cached work unit intent
* chore(ingest): verify resumable source ingest self-heal
* test(ingest): cover final gate prune source path resolution
* fix(ingest): resolve final gate prune sources canonically
* fix: defer wiki ref cleanup out of stage 3
* test: cover non-cascading final gate join pruning
* test: cover intrinsic final gate source drops
* docs(spec): record implementation notes for resumable source ingest (#22)
* fix(ingest): prune dangling joins on untouched sources and stop storing cache patch text
- final gate: drop a dropped source's dangling join edges from every owner on
the connection, including untouched siblings the touched-scoped gate never
revisits, so a committed orphan join can't break SL queries
- work-unit cache: drop the stored patch text; replay re-derives the diff from
the before/after artifact snapshots, carrying each touched file only once
- scan enrichment: checkpoint recomputed embeddings before the kill-prone
relationship stage even when descriptions load from disk, using the
best-available description set so the manifest merge can't delete them
- sl: extract listSlSourceFiles so the final gate and resolveSlSourceFile
share one listing path
* fix(scan): accept relationships mode in enrichment state metadata
Listing run stages after a relationships-mode scan threw "Invalid scan
enrichment cache metadata" because the parser hand-enumerated only the
structural/enriched modes while a relationships scan persists its stage with
mode 'relationships'. Derive the mode and stage allowlists from the canonical
KTX_SCAN_MODES and KTX_SCAN_ENRICHMENT_STAGES registries so the runtime check
cannot drift from the type again.
---------
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Andrey Avtomonov
GitHub
parent
46df7f3b24
commit
67a69dba8b
41 changed files with 4303 additions and 1599 deletions
64
packages/cli/src/context/cache/content-result-cache.ts
vendored
Normal file
64
packages/cli/src/context/cache/content-result-cache.ts
vendored
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
import { createHash } from 'node:crypto';
|
||||
|
||||
type ContentCacheMetadata = Record<string, unknown>;
|
||||
|
||||
export interface ContentResultCacheLookup {
|
||||
namespace: string;
|
||||
scopeKey: string;
|
||||
inputHash: string;
|
||||
}
|
||||
|
||||
export interface ContentResultCacheCompleted<TOutput = unknown> extends ContentResultCacheLookup {
|
||||
runId: string;
|
||||
status: 'completed';
|
||||
output: TOutput;
|
||||
errorMessage: null;
|
||||
metadata: ContentCacheMetadata;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface ContentResultCacheFailed extends ContentResultCacheLookup {
|
||||
runId: string;
|
||||
status: 'failed';
|
||||
output: null;
|
||||
errorMessage: string;
|
||||
metadata: ContentCacheMetadata;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export type ContentResultCacheRecord<TOutput = unknown> =
|
||||
| ContentResultCacheCompleted<TOutput>
|
||||
| ContentResultCacheFailed;
|
||||
|
||||
export interface ContentResultCache {
|
||||
findCompletedResult<TOutput = unknown>(
|
||||
input: ContentResultCacheLookup,
|
||||
): Promise<ContentResultCacheCompleted<TOutput> | null>;
|
||||
findLatestCompletedResult(input: {
|
||||
namespace: string;
|
||||
scopeKey: string;
|
||||
}): Promise<ContentResultCacheCompleted | null>;
|
||||
saveCompletedResult<TOutput = unknown>(
|
||||
input: Omit<ContentResultCacheCompleted<TOutput>, 'status' | 'errorMessage'>,
|
||||
): Promise<void>;
|
||||
saveFailedResult(input: Omit<ContentResultCacheFailed, 'status' | 'output'>): Promise<void>;
|
||||
deleteResult(input: ContentResultCacheLookup): Promise<void>;
|
||||
listRunResults(runId: string): Promise<ContentResultCacheRecord[]>;
|
||||
}
|
||||
|
||||
function stableJson(value: unknown): string {
|
||||
if (Array.isArray(value)) {
|
||||
return `[${value.map(stableJson).join(',')}]`;
|
||||
}
|
||||
if (value && typeof value === 'object') {
|
||||
const entries = Object.entries(value as Record<string, unknown>).sort(([left], [right]) =>
|
||||
left.localeCompare(right),
|
||||
);
|
||||
return `{${entries.map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(',')}}`;
|
||||
}
|
||||
return JSON.stringify(value) ?? 'undefined';
|
||||
}
|
||||
|
||||
export function stableContentHash(value: unknown): string {
|
||||
return createHash('sha256').update(stableJson(value)).digest('hex');
|
||||
}
|
||||
281
packages/cli/src/context/cache/sqlite-content-result-cache.ts
vendored
Normal file
281
packages/cli/src/context/cache/sqlite-content-result-cache.ts
vendored
Normal file
|
|
@ -0,0 +1,281 @@
|
|||
import { mkdirSync } from 'node:fs';
|
||||
import { dirname } from 'node:path';
|
||||
import Database from 'better-sqlite3';
|
||||
import type {
|
||||
ContentResultCache,
|
||||
ContentResultCacheCompleted,
|
||||
ContentResultCacheFailed,
|
||||
ContentResultCacheLookup,
|
||||
ContentResultCacheRecord,
|
||||
} from './content-result-cache.js';
|
||||
|
||||
export interface SqliteContentResultCacheOptions {
|
||||
dbPath: string;
|
||||
}
|
||||
|
||||
interface ResultRow {
|
||||
run_id: string;
|
||||
namespace: string;
|
||||
scope_key: string;
|
||||
input_hash: string;
|
||||
status: 'completed' | 'failed';
|
||||
output_json: string | null;
|
||||
error_message: string | null;
|
||||
metadata_json: string;
|
||||
updated_at: string;
|
||||
}
|
||||
|
||||
const RESULTS_TABLE = 'local_content_results';
|
||||
const RESULTS_PRIMARY_KEY = ['namespace', 'scope_key', 'input_hash'] as const;
|
||||
|
||||
function isSafeRunId(runId: string): boolean {
|
||||
return /^[a-zA-Z0-9][a-zA-Z0-9_.-]*$/.test(runId);
|
||||
}
|
||||
|
||||
function parseResultRow<TOutput = unknown>(row: ResultRow): ContentResultCacheRecord<TOutput> {
|
||||
const base = {
|
||||
runId: row.run_id,
|
||||
namespace: row.namespace,
|
||||
scopeKey: row.scope_key,
|
||||
inputHash: row.input_hash,
|
||||
metadata: JSON.parse(row.metadata_json || '{}') as Record<string, unknown>,
|
||||
updatedAt: row.updated_at,
|
||||
};
|
||||
if (row.status === 'completed') {
|
||||
return {
|
||||
...base,
|
||||
status: 'completed',
|
||||
output: JSON.parse(row.output_json ?? 'null') as TOutput,
|
||||
errorMessage: null,
|
||||
};
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
status: 'failed',
|
||||
output: null,
|
||||
errorMessage: row.error_message ?? 'Unknown content result failure',
|
||||
};
|
||||
}
|
||||
|
||||
export class SqliteContentResultCache implements ContentResultCache {
|
||||
private readonly db: Database.Database;
|
||||
|
||||
constructor(options: SqliteContentResultCacheOptions) {
|
||||
mkdirSync(dirname(options.dbPath), { recursive: true });
|
||||
this.db = new Database(options.dbPath);
|
||||
this.db.pragma('journal_mode = WAL');
|
||||
this.db.exec('DROP TABLE IF EXISTS local_scan_enrichment_stages');
|
||||
this.dropResultsTableIfPrimaryKeyDiffers();
|
||||
this.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS local_content_results (
|
||||
run_id TEXT NOT NULL,
|
||||
namespace TEXT NOT NULL,
|
||||
scope_key TEXT NOT NULL,
|
||||
input_hash TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
output_json TEXT,
|
||||
error_message TEXT,
|
||||
metadata_json TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (namespace, scope_key, input_hash)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS local_content_results_lookup_idx
|
||||
ON local_content_results (namespace, scope_key, input_hash, updated_at);
|
||||
CREATE INDEX IF NOT EXISTS local_content_results_run_idx
|
||||
ON local_content_results (run_id, updated_at, namespace);
|
||||
`);
|
||||
}
|
||||
|
||||
private dropResultsTableIfPrimaryKeyDiffers(): void {
|
||||
const columns = this.db.prepare(`PRAGMA table_info(${RESULTS_TABLE})`).all() as Array<{
|
||||
name: string;
|
||||
pk: number;
|
||||
}>;
|
||||
if (columns.length === 0) {
|
||||
return;
|
||||
}
|
||||
const primaryKey = columns
|
||||
.filter((column) => column.pk > 0)
|
||||
.sort((left, right) => left.pk - right.pk)
|
||||
.map((column) => column.name);
|
||||
const matches =
|
||||
primaryKey.length === RESULTS_PRIMARY_KEY.length &&
|
||||
primaryKey.every((name, index) => name === RESULTS_PRIMARY_KEY[index]);
|
||||
if (!matches) {
|
||||
this.db.exec(`DROP TABLE ${RESULTS_TABLE}`);
|
||||
}
|
||||
}
|
||||
|
||||
async findCompletedResult<TOutput = unknown>(
|
||||
input: ContentResultCacheLookup,
|
||||
): Promise<ContentResultCacheCompleted<TOutput> | null> {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_content_results
|
||||
WHERE namespace = ?
|
||||
AND scope_key = ?
|
||||
AND input_hash = ?
|
||||
AND status = 'completed'
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 1
|
||||
`,
|
||||
)
|
||||
.get(input.namespace, input.scopeKey, input.inputHash) as ResultRow | undefined;
|
||||
if (!row) {
|
||||
return null;
|
||||
}
|
||||
const parsed = parseResultRow<TOutput>(row);
|
||||
return parsed.status === 'completed' ? parsed : null;
|
||||
}
|
||||
|
||||
async findLatestCompletedResult(input: {
|
||||
namespace: string;
|
||||
scopeKey: string;
|
||||
}): Promise<ContentResultCacheCompleted | null> {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_content_results
|
||||
WHERE namespace = ?
|
||||
AND scope_key = ?
|
||||
AND status = 'completed'
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 1
|
||||
`,
|
||||
)
|
||||
.get(input.namespace, input.scopeKey) as ResultRow | undefined;
|
||||
if (!row) {
|
||||
return null;
|
||||
}
|
||||
const parsed = parseResultRow(row);
|
||||
return parsed.status === 'completed' ? parsed : null;
|
||||
}
|
||||
|
||||
async saveCompletedResult<TOutput = unknown>(
|
||||
input: Omit<ContentResultCacheCompleted<TOutput>, 'status' | 'errorMessage'>,
|
||||
): Promise<void> {
|
||||
this.db
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO local_content_results (
|
||||
run_id,
|
||||
namespace,
|
||||
scope_key,
|
||||
input_hash,
|
||||
status,
|
||||
output_json,
|
||||
error_message,
|
||||
metadata_json,
|
||||
updated_at
|
||||
)
|
||||
VALUES (
|
||||
@runId,
|
||||
@namespace,
|
||||
@scopeKey,
|
||||
@inputHash,
|
||||
'completed',
|
||||
@outputJson,
|
||||
NULL,
|
||||
@metadataJson,
|
||||
@updatedAt
|
||||
)
|
||||
ON CONFLICT(namespace, scope_key, input_hash) DO UPDATE SET
|
||||
run_id = excluded.run_id,
|
||||
status = excluded.status,
|
||||
output_json = excluded.output_json,
|
||||
error_message = excluded.error_message,
|
||||
metadata_json = excluded.metadata_json,
|
||||
updated_at = excluded.updated_at
|
||||
`,
|
||||
)
|
||||
.run({
|
||||
runId: input.runId,
|
||||
namespace: input.namespace,
|
||||
scopeKey: input.scopeKey,
|
||||
inputHash: input.inputHash,
|
||||
outputJson: JSON.stringify(input.output),
|
||||
metadataJson: JSON.stringify(input.metadata),
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
}
|
||||
|
||||
async saveFailedResult(input: Omit<ContentResultCacheFailed, 'status' | 'output'>): Promise<void> {
|
||||
this.db
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO local_content_results (
|
||||
run_id,
|
||||
namespace,
|
||||
scope_key,
|
||||
input_hash,
|
||||
status,
|
||||
output_json,
|
||||
error_message,
|
||||
metadata_json,
|
||||
updated_at
|
||||
)
|
||||
VALUES (
|
||||
@runId,
|
||||
@namespace,
|
||||
@scopeKey,
|
||||
@inputHash,
|
||||
'failed',
|
||||
NULL,
|
||||
@errorMessage,
|
||||
@metadataJson,
|
||||
@updatedAt
|
||||
)
|
||||
ON CONFLICT(namespace, scope_key, input_hash) DO UPDATE SET
|
||||
run_id = excluded.run_id,
|
||||
status = excluded.status,
|
||||
output_json = excluded.output_json,
|
||||
error_message = excluded.error_message,
|
||||
metadata_json = excluded.metadata_json,
|
||||
updated_at = excluded.updated_at
|
||||
`,
|
||||
)
|
||||
.run({
|
||||
runId: input.runId,
|
||||
namespace: input.namespace,
|
||||
scopeKey: input.scopeKey,
|
||||
inputHash: input.inputHash,
|
||||
errorMessage: input.errorMessage,
|
||||
metadataJson: JSON.stringify(input.metadata),
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
}
|
||||
|
||||
async deleteResult(input: ContentResultCacheLookup): Promise<void> {
|
||||
this.db
|
||||
.prepare(
|
||||
`
|
||||
DELETE FROM local_content_results
|
||||
WHERE namespace = ?
|
||||
AND scope_key = ?
|
||||
AND input_hash = ?
|
||||
`,
|
||||
)
|
||||
.run(input.namespace, input.scopeKey, input.inputHash);
|
||||
}
|
||||
|
||||
async listRunResults(runId: string): Promise<ContentResultCacheRecord[]> {
|
||||
if (!isSafeRunId(runId)) {
|
||||
return [];
|
||||
}
|
||||
const rows = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_content_results
|
||||
WHERE run_id = ?
|
||||
ORDER BY updated_at ASC, namespace ASC
|
||||
`,
|
||||
)
|
||||
.all(runId) as ResultRow[];
|
||||
return rows.map((row) => parseResultRow(row));
|
||||
}
|
||||
}
|
||||
|
|
@ -3,7 +3,7 @@ import type { TouchedSlSource } from '../../context/tools/touched-sl-sources.js'
|
|||
import type { KnowledgeWikiService } from '../../context/wiki/knowledge-wiki.service.js';
|
||||
import { findMissingWikiRefs } from '../wiki/wiki-ref-validation.js';
|
||||
import type { WuValidationResult } from './stages/validate-wu-sources.js';
|
||||
import { findInvalidWikiBodyRefs } from './wiki-body-refs.js';
|
||||
import { findInvalidWikiBodyRefIssues, type WikiBodyRefIssue } from './wiki-body-refs.js';
|
||||
|
||||
export interface FinalArtifactGateInput {
|
||||
connectionIds: string[];
|
||||
|
|
@ -21,6 +21,31 @@ export interface ProvenanceRawPathValidationInput {
|
|||
deletedRawPaths: Set<string>;
|
||||
}
|
||||
|
||||
export type FinalArtifactGateFinding =
|
||||
| { kind: 'invalid_source'; connectionId: string; sourceName: string; errors: string[] }
|
||||
| {
|
||||
kind: 'missing_join_target';
|
||||
ownerConnectionId: string;
|
||||
ownerSourceName: string;
|
||||
targetSourceName: string;
|
||||
message: string;
|
||||
}
|
||||
| { kind: 'missing_wiki_ref'; pageKey: string; targetPageKey: string; message: string }
|
||||
| {
|
||||
kind: 'missing_wiki_sl_ref';
|
||||
pageKey: string;
|
||||
ref: string;
|
||||
sourceName: string;
|
||||
entityName: string | null;
|
||||
message: string;
|
||||
}
|
||||
| WikiBodyRefIssue;
|
||||
|
||||
export interface FinalArtifactGateResult {
|
||||
ok: boolean;
|
||||
findings: FinalArtifactGateFinding[];
|
||||
}
|
||||
|
||||
function normalizeRawPath(path: string): string {
|
||||
return path.replace(/\\/g, '/').replace(/^\/+/, '');
|
||||
}
|
||||
|
|
@ -40,8 +65,8 @@ function slEntityNames(source: Awaited<ReturnType<SemanticLayerService['loadAllS
|
|||
]);
|
||||
}
|
||||
|
||||
async function validateWikiSlRefs(input: FinalArtifactGateInput): Promise<string[]> {
|
||||
const errors: string[] = [];
|
||||
async function validateWikiSlRefs(input: FinalArtifactGateInput): Promise<FinalArtifactGateFinding[]> {
|
||||
const findings: FinalArtifactGateFinding[] = [];
|
||||
const sourcesByConnection = new Map<string, Awaited<ReturnType<SemanticLayerService['loadAllSources']>>['sources']>();
|
||||
for (const connectionId of input.connectionIds) {
|
||||
const { sources } = await input.semanticLayerService.loadAllSources(connectionId);
|
||||
|
|
@ -64,19 +89,33 @@ async function validateWikiSlRefs(input: FinalArtifactGateInput): Promise<string
|
|||
}
|
||||
}
|
||||
if (!source) {
|
||||
errors.push(`${pageKey}: unknown sl_refs entry ${ref}`);
|
||||
findings.push({
|
||||
kind: 'missing_wiki_sl_ref',
|
||||
pageKey,
|
||||
ref,
|
||||
sourceName: parsed.sourceName,
|
||||
entityName: parsed.entityName,
|
||||
message: `${pageKey}: unknown sl_refs entry ${ref}`,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (parsed.entityName && !slEntityNames(source).has(parsed.entityName)) {
|
||||
errors.push(`${pageKey}: unknown sl_refs entity ${ref}`);
|
||||
findings.push({
|
||||
kind: 'missing_wiki_sl_ref',
|
||||
pageKey,
|
||||
ref,
|
||||
sourceName: parsed.sourceName,
|
||||
entityName: parsed.entityName,
|
||||
message: `${pageKey}: unknown sl_refs entity ${ref}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
return errors;
|
||||
return findings;
|
||||
}
|
||||
|
||||
async function validateWikiRefs(input: FinalArtifactGateInput): Promise<string[]> {
|
||||
const dangling: string[] = [];
|
||||
async function validateWikiRefs(input: FinalArtifactGateInput): Promise<FinalArtifactGateFinding[]> {
|
||||
const findings: FinalArtifactGateFinding[] = [];
|
||||
for (const pageKey of input.changedWikiPageKeys) {
|
||||
const page = await input.wikiService.readPage('GLOBAL', null, pageKey);
|
||||
if (!page) {
|
||||
|
|
@ -91,33 +130,82 @@ async function validateWikiRefs(input: FinalArtifactGateInput): Promise<string[]
|
|||
content: page.content,
|
||||
});
|
||||
for (const missingRef of missingRefs) {
|
||||
dangling.push(`${pageKey} -> ${missingRef}`);
|
||||
findings.push({
|
||||
kind: 'missing_wiki_ref',
|
||||
pageKey,
|
||||
targetPageKey: missingRef,
|
||||
message: `${pageKey} -> ${missingRef}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
return dangling;
|
||||
return findings;
|
||||
}
|
||||
|
||||
export async function validateFinalIngestArtifacts(input: FinalArtifactGateInput): Promise<void> {
|
||||
export function formatFinalArtifactGateFindings(findings: FinalArtifactGateFinding[]): string {
|
||||
const errors = findings.map((finding) => {
|
||||
if (finding.kind === 'invalid_source') {
|
||||
return `semantic-layer validation failed for ${finding.connectionId}:${finding.sourceName}: ${finding.errors.join('; ')}`;
|
||||
}
|
||||
if (finding.kind === 'missing_wiki_ref') {
|
||||
return `wiki reference targets missing page: ${finding.message}`;
|
||||
}
|
||||
return finding.message;
|
||||
});
|
||||
return `final artifact gates failed:\n${errors.join('\n')}`;
|
||||
}
|
||||
|
||||
export function isFinalArtifactGateFindingPruneable(finding: FinalArtifactGateFinding): boolean {
|
||||
switch (finding.kind) {
|
||||
case 'invalid_source':
|
||||
case 'missing_join_target':
|
||||
case 'missing_wiki_ref':
|
||||
case 'missing_wiki_sl_ref':
|
||||
case 'missing_wiki_body_sl_entity':
|
||||
case 'missing_wiki_body_sl_source':
|
||||
case 'missing_wiki_body_table':
|
||||
return true;
|
||||
default: {
|
||||
const exhaustive: never = finding;
|
||||
return exhaustive;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function validateFinalIngestArtifacts(input: FinalArtifactGateInput): Promise<FinalArtifactGateResult> {
|
||||
// Join-neighbor expansion happens inside validateTouchedSources so work-unit
|
||||
// validation and this gate check the same set — a source that passes one
|
||||
// passes the other.
|
||||
const validation = await input.validateTouchedSources(input.touchedSlSources);
|
||||
const errors: string[] = validation.invalidSources.map(
|
||||
(invalid) => `semantic-layer validation failed for ${invalid.source}: ${invalid.errors.join('; ')}`,
|
||||
);
|
||||
errors.push(...(await validateWikiSlRefs(input)));
|
||||
const danglingWikiRefs = await validateWikiRefs(input);
|
||||
if (danglingWikiRefs.length > 0) {
|
||||
errors.push(`wiki references target missing page(s): ${danglingWikiRefs.join(', ')}`);
|
||||
const findings: FinalArtifactGateFinding[] = [];
|
||||
for (const invalid of validation.invalidSources) {
|
||||
const [connectionId = '', sourceName = ''] = invalid.source.split(':', 2);
|
||||
const issues = invalid.issues ?? invalid.errors.map((message) => ({ kind: 'source_validation' as const, message }));
|
||||
const sourceErrors = issues.filter((issue) => issue.kind === 'source_validation').map((issue) => issue.message);
|
||||
if (sourceErrors.length > 0) {
|
||||
findings.push({ kind: 'invalid_source', connectionId, sourceName, errors: sourceErrors });
|
||||
}
|
||||
for (const issue of issues) {
|
||||
if (issue.kind === 'missing_join_target') {
|
||||
findings.push({
|
||||
kind: 'missing_join_target',
|
||||
ownerConnectionId: connectionId,
|
||||
ownerSourceName: sourceName,
|
||||
targetSourceName: issue.targetSourceName,
|
||||
message: issue.message,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
findings.push(...(await validateWikiSlRefs(input)));
|
||||
findings.push(...(await validateWikiRefs(input)));
|
||||
|
||||
for (const pageKey of input.changedWikiPageKeys) {
|
||||
const page = await input.wikiService.readPage('GLOBAL', null, pageKey);
|
||||
if (!page) {
|
||||
continue;
|
||||
}
|
||||
errors.push(
|
||||
...(await findInvalidWikiBodyRefs({
|
||||
findings.push(
|
||||
...(await findInvalidWikiBodyRefIssues({
|
||||
pageKey,
|
||||
body: page.content,
|
||||
visibleConnectionIds: input.connectionIds,
|
||||
|
|
@ -130,9 +218,7 @@ export async function validateFinalIngestArtifacts(input: FinalArtifactGateInput
|
|||
);
|
||||
}
|
||||
|
||||
if (errors.length > 0) {
|
||||
throw new Error(`final artifact gates failed:\n${errors.join('\n')}`);
|
||||
}
|
||||
return { ok: findings.length === 0, findings };
|
||||
}
|
||||
|
||||
export function validateProvenanceRawPaths(input: ProvenanceRawPathValidationInput): void {
|
||||
|
|
|
|||
330
packages/cli/src/context/ingest/final-gate-prune.ts
Normal file
330
packages/cli/src/context/ingest/final-gate-prune.ts
Normal file
|
|
@ -0,0 +1,330 @@
|
|||
import YAML from 'yaml';
|
||||
import type { KtxFileStorePort } from '../core/file-store.js';
|
||||
import { listSlSourceFiles, resolveSlSourceFile, slSourceNameForFile } from '../sl/source-files.js';
|
||||
import type { KnowledgeWikiService } from '../wiki/knowledge-wiki.service.js';
|
||||
import type { FinalArtifactGateFinding } from './artifact-gates.js';
|
||||
import type { IngestTraceWriter } from './ingest-trace.js';
|
||||
|
||||
type FinalGatePrunedReferenceKind = 'join' | 'wiki_ref' | 'wiki_sl_ref' | 'wiki_body_ref';
|
||||
type SemanticLayerFileStore = Pick<KtxFileStorePort, 'readFile' | 'writeFile' | 'deleteFile' | 'listFiles'>;
|
||||
|
||||
interface ResolvedYamlSource {
|
||||
path: string;
|
||||
source: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface FinalGatePrunedReference {
|
||||
kind: FinalGatePrunedReferenceKind;
|
||||
artifact: string;
|
||||
removedRef: string;
|
||||
absentTarget: string;
|
||||
}
|
||||
|
||||
export interface FinalGateDroppedSource {
|
||||
connectionId: string;
|
||||
sourceName: string;
|
||||
reason: string;
|
||||
}
|
||||
|
||||
export interface FinalGatePruneResult {
|
||||
prunedReferences: FinalGatePrunedReference[];
|
||||
droppedSources: FinalGateDroppedSource[];
|
||||
}
|
||||
|
||||
interface PruneInput {
|
||||
workdir: string;
|
||||
semanticLayerFiles: SemanticLayerFileStore;
|
||||
findings: FinalArtifactGateFinding[];
|
||||
droppedSources: FinalGateDroppedSource[];
|
||||
trace: IngestTraceWriter;
|
||||
author: { name: string; email: string };
|
||||
wikiService?: KnowledgeWikiService;
|
||||
}
|
||||
|
||||
async function resolveYamlSource(
|
||||
fileStore: SemanticLayerFileStore,
|
||||
connectionId: string,
|
||||
sourceName: string,
|
||||
): Promise<ResolvedYamlSource | null> {
|
||||
const file = await resolveSlSourceFile(fileStore, connectionId, sourceName);
|
||||
if (!file) {
|
||||
return null;
|
||||
}
|
||||
const parsed = YAML.parse(file.content);
|
||||
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
|
||||
throw new Error(`${file.path}: expected semantic-layer source YAML object`);
|
||||
}
|
||||
return { path: file.path, source: parsed as Record<string, unknown> };
|
||||
}
|
||||
|
||||
async function writeYamlSource(input: {
|
||||
fileStore: SemanticLayerFileStore;
|
||||
path: string;
|
||||
source: Record<string, unknown>;
|
||||
author: { name: string; email: string };
|
||||
}): Promise<void> {
|
||||
await input.fileStore.writeFile(
|
||||
input.path,
|
||||
YAML.stringify(input.source, { indent: 2, lineWidth: 0, version: '1.1' }),
|
||||
input.author.name,
|
||||
input.author.email,
|
||||
`Prune dangling joins from ${input.path}`,
|
||||
{ skipLock: true },
|
||||
);
|
||||
}
|
||||
|
||||
function removeInlineToken(content: string, rawToken: string): string {
|
||||
return content.replaceAll(`\`${rawToken}\``, '').replace(/[ \t]+([.,;:!?])/g, '$1');
|
||||
}
|
||||
|
||||
function escapeRegExp(value: string): string {
|
||||
return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
|
||||
}
|
||||
|
||||
function removeWikiRefToken(content: string, targetPageKey: string): string {
|
||||
const pattern = new RegExp(`\\[\\[\\s*${escapeRegExp(targetPageKey)}(?:\\|[^\\]\\n]+)?\\s*\\]\\]`, 'g');
|
||||
return content.replace(pattern, '').replace(/[ \t]+([.,;:!?])/g, '$1');
|
||||
}
|
||||
|
||||
function wikiBodyAbsentTarget(finding: FinalArtifactGateFinding): string {
|
||||
if (finding.kind === 'missing_wiki_body_table') {
|
||||
return finding.tableRef;
|
||||
}
|
||||
if (finding.kind === 'missing_wiki_body_sl_source') {
|
||||
return finding.sourceName;
|
||||
}
|
||||
if (finding.kind === 'missing_wiki_body_sl_entity') {
|
||||
return `${finding.sourceName}.${finding.entityName}`;
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
/** Remove every join whose target matches `shouldRemove`, write the file back, and
|
||||
* emit one pruned-reference record per distinct removed target. */
|
||||
async function pruneJoinsFromSource(input: {
|
||||
fileStore: SemanticLayerFileStore;
|
||||
connectionId: string;
|
||||
ownerSourceName: string;
|
||||
resolved: ResolvedYamlSource;
|
||||
shouldRemove: (target: string) => boolean;
|
||||
author: { name: string; email: string };
|
||||
trace: IngestTraceWriter;
|
||||
}): Promise<FinalGatePrunedReference[]> {
|
||||
if (!Array.isArray(input.resolved.source.joins)) {
|
||||
return [];
|
||||
}
|
||||
const removed = new Set<string>();
|
||||
const nextJoins = input.resolved.source.joins.filter((entry) => {
|
||||
const to = entry && typeof entry === 'object' && 'to' in entry ? (entry as { to: unknown }).to : undefined;
|
||||
if (typeof to === 'string' && input.shouldRemove(to)) {
|
||||
removed.add(to);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
if (removed.size === 0) {
|
||||
return [];
|
||||
}
|
||||
input.resolved.source.joins = nextJoins;
|
||||
await writeYamlSource({
|
||||
fileStore: input.fileStore,
|
||||
path: input.resolved.path,
|
||||
source: input.resolved.source,
|
||||
author: input.author,
|
||||
});
|
||||
const records: FinalGatePrunedReference[] = [];
|
||||
for (const target of removed) {
|
||||
const record = {
|
||||
kind: 'join' as const,
|
||||
artifact: `semantic-layer/${input.connectionId}/${input.ownerSourceName}`,
|
||||
removedRef: target,
|
||||
absentTarget: target,
|
||||
};
|
||||
records.push(record);
|
||||
await input.trace.event('info', 'final_gates', 'final_gate_reference_pruned', record);
|
||||
}
|
||||
return records;
|
||||
}
|
||||
|
||||
export async function pruneFinalGateFindings(input: PruneInput): Promise<FinalGatePruneResult> {
|
||||
const droppedSources = [...input.droppedSources];
|
||||
const prunedReferences: FinalGatePrunedReference[] = [];
|
||||
const droppedKey = new Set(droppedSources.map((source) => `${source.connectionId}:${source.sourceName}`));
|
||||
|
||||
for (const finding of input.findings) {
|
||||
if (finding.kind !== 'invalid_source') {
|
||||
continue;
|
||||
}
|
||||
const key = `${finding.connectionId}:${finding.sourceName}`;
|
||||
if (droppedKey.has(key)) {
|
||||
continue;
|
||||
}
|
||||
const file = await resolveSlSourceFile(input.semanticLayerFiles, finding.connectionId, finding.sourceName);
|
||||
if (!file) {
|
||||
continue;
|
||||
}
|
||||
const deleted = await input.semanticLayerFiles.deleteFile(
|
||||
file.path,
|
||||
input.author.name,
|
||||
input.author.email,
|
||||
`Drop invalid source ${finding.connectionId}:${finding.sourceName}`,
|
||||
{ skipLock: true },
|
||||
);
|
||||
if (!deleted) {
|
||||
continue;
|
||||
}
|
||||
const dropped = {
|
||||
connectionId: finding.connectionId,
|
||||
sourceName: finding.sourceName,
|
||||
reason: finding.errors.join('; '),
|
||||
};
|
||||
droppedSources.push(dropped);
|
||||
droppedKey.add(key);
|
||||
await input.trace.event('info', 'final_gates', 'final_gate_source_dropped', dropped);
|
||||
}
|
||||
|
||||
// A dropped node can leave a join dangling on any owner — including sources
|
||||
// untouched by this run, which the touched-scoped gate (and the confirm gate
|
||||
// after it) never revisit. Prune those edges directly (D5), or the committed
|
||||
// orphan join breaks every SL query on the connection.
|
||||
const droppedByConnection = new Map<string, Set<string>>();
|
||||
for (const dropped of droppedSources) {
|
||||
const names = droppedByConnection.get(dropped.connectionId) ?? new Set<string>();
|
||||
names.add(dropped.sourceName);
|
||||
droppedByConnection.set(dropped.connectionId, names);
|
||||
}
|
||||
for (const [connectionId, droppedNames] of droppedByConnection) {
|
||||
for (const file of await listSlSourceFiles(input.semanticLayerFiles, connectionId)) {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = YAML.parse(file.content);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
|
||||
continue;
|
||||
}
|
||||
prunedReferences.push(
|
||||
...(await pruneJoinsFromSource({
|
||||
fileStore: input.semanticLayerFiles,
|
||||
connectionId,
|
||||
ownerSourceName: slSourceNameForFile(file.path, file.content),
|
||||
resolved: { path: file.path, source: parsed as Record<string, unknown> },
|
||||
shouldRemove: (target) => droppedNames.has(target),
|
||||
author: input.author,
|
||||
trace: input.trace,
|
||||
})),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
for (const finding of input.findings) {
|
||||
if (finding.kind !== 'missing_join_target') {
|
||||
continue;
|
||||
}
|
||||
const resolved = await resolveYamlSource(
|
||||
input.semanticLayerFiles,
|
||||
finding.ownerConnectionId,
|
||||
finding.ownerSourceName,
|
||||
);
|
||||
if (!resolved) {
|
||||
continue;
|
||||
}
|
||||
prunedReferences.push(
|
||||
...(await pruneJoinsFromSource({
|
||||
fileStore: input.semanticLayerFiles,
|
||||
connectionId: finding.ownerConnectionId,
|
||||
ownerSourceName: finding.ownerSourceName,
|
||||
resolved,
|
||||
shouldRemove: (target) => target === finding.targetSourceName,
|
||||
author: input.author,
|
||||
trace: input.trace,
|
||||
})),
|
||||
);
|
||||
}
|
||||
|
||||
const wikiFindings = input.findings.filter(
|
||||
(finding) =>
|
||||
finding.kind === 'missing_wiki_ref' ||
|
||||
finding.kind === 'missing_wiki_sl_ref' ||
|
||||
finding.kind === 'missing_wiki_body_sl_source' ||
|
||||
finding.kind === 'missing_wiki_body_sl_entity' ||
|
||||
finding.kind === 'missing_wiki_body_table',
|
||||
);
|
||||
const pageKeys = [...new Set(wikiFindings.map((finding) => finding.pageKey))].sort();
|
||||
for (const pageKey of pageKeys) {
|
||||
const page = input.wikiService ? await input.wikiService.readPage('GLOBAL', null, pageKey) : null;
|
||||
if (!page) {
|
||||
continue;
|
||||
}
|
||||
const frontmatter = { ...page.frontmatter };
|
||||
let content = page.content;
|
||||
let changed = false;
|
||||
for (const finding of wikiFindings.filter((candidate) => candidate.pageKey === pageKey)) {
|
||||
if (finding.kind === 'missing_wiki_ref') {
|
||||
const refs = Array.isArray(frontmatter.refs) ? frontmatter.refs.filter((ref) => ref !== finding.targetPageKey) : [];
|
||||
const nextContent = removeWikiRefToken(content, finding.targetPageKey);
|
||||
if ((Array.isArray(frontmatter.refs) && refs.length !== frontmatter.refs.length) || nextContent !== content) {
|
||||
if (Array.isArray(frontmatter.refs)) {
|
||||
frontmatter.refs = refs;
|
||||
}
|
||||
content = nextContent;
|
||||
changed = true;
|
||||
const record = {
|
||||
kind: 'wiki_ref' as const,
|
||||
artifact: `wiki/global/${pageKey}`,
|
||||
removedRef: finding.targetPageKey,
|
||||
absentTarget: finding.targetPageKey,
|
||||
};
|
||||
prunedReferences.push(record);
|
||||
await input.trace.event('info', 'final_gates', 'final_gate_reference_pruned', record);
|
||||
}
|
||||
} else if (finding.kind === 'missing_wiki_sl_ref') {
|
||||
const slRefs = Array.isArray(frontmatter.sl_refs)
|
||||
? frontmatter.sl_refs.filter((ref) => ref !== finding.ref)
|
||||
: [];
|
||||
if (Array.isArray(frontmatter.sl_refs) && slRefs.length !== frontmatter.sl_refs.length) {
|
||||
frontmatter.sl_refs = slRefs;
|
||||
changed = true;
|
||||
const record = {
|
||||
kind: 'wiki_sl_ref' as const,
|
||||
artifact: `wiki/global/${pageKey}`,
|
||||
removedRef: finding.ref,
|
||||
absentTarget: finding.sourceName,
|
||||
};
|
||||
prunedReferences.push(record);
|
||||
await input.trace.event('info', 'final_gates', 'final_gate_reference_pruned', record);
|
||||
}
|
||||
} else {
|
||||
const nextContent = removeInlineToken(content, finding.rawToken);
|
||||
if (nextContent !== content) {
|
||||
content = nextContent;
|
||||
changed = true;
|
||||
const record = {
|
||||
kind: 'wiki_body_ref' as const,
|
||||
artifact: `wiki/global/${pageKey}`,
|
||||
removedRef: finding.rawToken,
|
||||
absentTarget: wikiBodyAbsentTarget(finding),
|
||||
};
|
||||
prunedReferences.push(record);
|
||||
await input.trace.event('info', 'final_gates', 'final_gate_reference_pruned', record);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (changed && input.wikiService) {
|
||||
await input.wikiService.writePage(
|
||||
'GLOBAL',
|
||||
null,
|
||||
pageKey,
|
||||
frontmatter,
|
||||
content,
|
||||
input.author.name,
|
||||
input.author.email,
|
||||
`Prune dangling refs from ${pageKey}`,
|
||||
{ skipLock: true },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { prunedReferences, droppedSources };
|
||||
}
|
||||
|
|
@ -1,136 +0,0 @@
|
|||
import { z } from 'zod';
|
||||
import type { AgentRunnerPort, KtxRuntimeToolSet } from '../../context/llm/runtime-port.js';
|
||||
import type { ConstrainedRepairResult, RepairVerification } from './constrained-repair.js';
|
||||
import { runConstrainedRepairLoop } from './constrained-repair.js';
|
||||
import type { IngestTraceWriter } from './ingest-trace.js';
|
||||
|
||||
type FinalGateRepairKind = 'patch_semantic_gate' | 'final_artifact_gate';
|
||||
|
||||
export type FinalGateRepairResult = ConstrainedRepairResult;
|
||||
|
||||
export interface RepairFinalGateFailureInput {
|
||||
agentRunner: AgentRunnerPort;
|
||||
workdir: string;
|
||||
gateError: string;
|
||||
allowedPaths: string[];
|
||||
trace: IngestTraceWriter;
|
||||
repairKind: FinalGateRepairKind;
|
||||
/**
|
||||
* Re-runs the failed gate against the current worktree. The repair counts
|
||||
* as successful only when this passes — editing files is not the success
|
||||
* signal.
|
||||
*/
|
||||
verify(changedPaths: string[]): Promise<RepairVerification>;
|
||||
maxAttempts?: number;
|
||||
stepBudget?: number;
|
||||
abortSignal?: AbortSignal;
|
||||
}
|
||||
|
||||
function buildGateRepairSystemPrompt(): string {
|
||||
return `<role>
|
||||
You repair one ktx isolated-diff artifact gate failure inside the integration worktree.
|
||||
</role>
|
||||
|
||||
<rules>
|
||||
- Use read_gate_error first.
|
||||
- Read only files exposed by read_repair_file.
|
||||
- Edit only paths exposed by write_repair_file.
|
||||
- Prefer the smallest text edit that makes the gate pass.
|
||||
- Preserve accepted work-unit, reconciliation, and deterministic projection content.
|
||||
- Do not invent warehouse facts, business definitions, or semantic-layer entities.
|
||||
- If the gate error requires choosing between conflicting facts without evidence, stop without editing.
|
||||
</rules>`;
|
||||
}
|
||||
|
||||
function buildGateRepairUserPrompt(input: {
|
||||
gateError: string;
|
||||
allowedPaths: string[];
|
||||
repairKind: FinalGateRepairKind;
|
||||
attempt: number;
|
||||
maxAttempts: number;
|
||||
previousFailure: string | null;
|
||||
}): string {
|
||||
const previousFailureBlock = input.previousFailure
|
||||
? `\nPrevious attempt did not pass the gate:\n${input.previousFailure}\n`
|
||||
: '';
|
||||
return `Repair isolated-diff artifact gates.
|
||||
|
||||
Repair kind: ${input.repairKind}
|
||||
Attempt: ${input.attempt} of ${input.maxAttempts}
|
||||
|
||||
Allowed files:
|
||||
${input.allowedPaths.map((path) => `- ${path}`).join('\n')}
|
||||
|
||||
Gate error:
|
||||
${input.gateError}
|
||||
${previousFailureBlock}
|
||||
Use read_gate_error first. Then inspect only the allowed files, write the
|
||||
minimal repaired content, and stop.`;
|
||||
}
|
||||
|
||||
function buildReadGateErrorTool(gateError: string): KtxRuntimeToolSet {
|
||||
return {
|
||||
read_gate_error: {
|
||||
name: 'read_gate_error',
|
||||
description: 'Read the artifact gate failure that must be repaired.',
|
||||
inputSchema: z.object({}),
|
||||
execute: async () => ({
|
||||
markdown: gateError,
|
||||
structured: { gateError },
|
||||
}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function finalGateRepairPaths(input: {
|
||||
changedWikiPageKeys: string[];
|
||||
// Resolved by the caller: SL filenames are derived labels, so the repair
|
||||
// allowlist must carry the real on-disk paths, not name-interpolated ones.
|
||||
touchedSlSourcePaths: string[];
|
||||
}): string[] {
|
||||
return [
|
||||
...new Set([
|
||||
...input.touchedSlSourcePaths,
|
||||
...input.changedWikiPageKeys.map((pageKey) => `wiki/global/${pageKey}.md`),
|
||||
]),
|
||||
].sort();
|
||||
}
|
||||
|
||||
export async function repairFinalGateFailure(
|
||||
input: RepairFinalGateFailureInput,
|
||||
): Promise<FinalGateRepairResult> {
|
||||
return runConstrainedRepairLoop({
|
||||
agentRunner: input.agentRunner,
|
||||
workdir: input.workdir,
|
||||
allowedPaths: input.allowedPaths,
|
||||
trace: input.trace,
|
||||
tracePhase: 'gate_repair',
|
||||
traceEventName: 'gate_repair',
|
||||
traceData: {
|
||||
repairKind: input.repairKind,
|
||||
gateError: input.gateError,
|
||||
},
|
||||
systemPrompt: buildGateRepairSystemPrompt(),
|
||||
buildUserPrompt: ({ attempt, maxAttempts, previousFailure }) =>
|
||||
buildGateRepairUserPrompt({
|
||||
gateError: input.gateError,
|
||||
allowedPaths: [...input.allowedPaths].sort(),
|
||||
repairKind: input.repairKind,
|
||||
attempt,
|
||||
maxAttempts,
|
||||
previousFailure,
|
||||
}),
|
||||
buildExtraTools: () => buildReadGateErrorTool(input.gateError),
|
||||
verify: input.verify,
|
||||
noChangeFailureReason: 'gate repair completed without editing an allowed path',
|
||||
telemetryTags: {
|
||||
operationName: 'ingest-isolated-diff-gate-repair',
|
||||
source: input.trace.context.sourceKey,
|
||||
jobId: input.trace.context.jobId,
|
||||
repairKind: input.repairKind,
|
||||
},
|
||||
maxAttempts: input.maxAttempts,
|
||||
stepBudget: input.stepBudget ?? 16,
|
||||
abortSignal: input.abortSignal,
|
||||
});
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -1,7 +1,6 @@
|
|||
import { readFile } from 'node:fs/promises';
|
||||
import type { GitService } from '../../../context/core/git.service.js';
|
||||
import type { RepairVerification } from '../constrained-repair.js';
|
||||
import type { FinalGateRepairResult } from '../final-gate-repair.js';
|
||||
import type { IngestTraceWriter } from '../ingest-trace.js';
|
||||
import { traceTimed } from '../ingest-trace.js';
|
||||
import { assertPatchAllowedForWorkUnit, parsePatchTouchedPaths } from './git-patch.js';
|
||||
|
|
@ -13,21 +12,18 @@ export type PatchIntegrationResult =
|
|||
commitSha: string;
|
||||
touchedPaths: string[];
|
||||
textualResolution?: TextualConflictResolutionResult;
|
||||
gateRepair?: FinalGateRepairResult;
|
||||
}
|
||||
| {
|
||||
status: 'textual_conflict';
|
||||
reason: string;
|
||||
touchedPaths: string[];
|
||||
textualResolution?: TextualConflictResolutionResult;
|
||||
gateRepair?: FinalGateRepairResult;
|
||||
}
|
||||
| {
|
||||
status: 'semantic_conflict';
|
||||
reason: string;
|
||||
touchedPaths: string[];
|
||||
textualResolution?: TextualConflictResolutionResult;
|
||||
gateRepair?: FinalGateRepairResult;
|
||||
};
|
||||
|
||||
export interface IntegrateWorkUnitPatchInput {
|
||||
|
|
@ -46,13 +42,6 @@ export interface IntegrateWorkUnitPatchInput {
|
|||
reason: string;
|
||||
verify(changedPaths: string[]): Promise<RepairVerification>;
|
||||
}): Promise<TextualConflictResolutionResult>;
|
||||
repairGateFailure?(input: {
|
||||
unitKey: string;
|
||||
patchPath: string;
|
||||
touchedPaths: string[];
|
||||
reason: string;
|
||||
verify(changedPaths: string[]): Promise<RepairVerification>;
|
||||
}): Promise<FinalGateRepairResult>;
|
||||
}
|
||||
|
||||
function errorMessage(error: unknown): string {
|
||||
|
|
@ -225,59 +214,6 @@ export async function integrateWorkUnitPatch(input: IntegrateWorkUnitPatchInput)
|
|||
reason,
|
||||
});
|
||||
|
||||
if (input.repairGateFailure) {
|
||||
const gateRepair = await input.repairGateFailure({
|
||||
unitKey: input.unitKey,
|
||||
patchPath: input.patchPath,
|
||||
touchedPaths,
|
||||
reason,
|
||||
verify: verifyAppliedTree,
|
||||
});
|
||||
|
||||
if (gateRepair.status === 'failed') {
|
||||
if (preApplyHead) {
|
||||
await input.integrationGit.resetHardTo(preApplyHead);
|
||||
}
|
||||
return {
|
||||
status: 'semantic_conflict',
|
||||
reason: gateRepair.reason,
|
||||
touchedPaths,
|
||||
gateRepair,
|
||||
};
|
||||
}
|
||||
|
||||
const commit = await input.integrationGit.commitFiles(
|
||||
gateRepair.changedPaths,
|
||||
`ingest: repair WorkUnit ${input.unitKey} gates`,
|
||||
input.author.name,
|
||||
input.author.email,
|
||||
);
|
||||
if (!commit.created) {
|
||||
if (preApplyHead) {
|
||||
await input.integrationGit.resetHardTo(preApplyHead);
|
||||
}
|
||||
return {
|
||||
status: 'semantic_conflict',
|
||||
reason: 'gate repair produced no committable changes',
|
||||
touchedPaths: gateRepair.changedPaths,
|
||||
gateRepair,
|
||||
};
|
||||
}
|
||||
|
||||
await input.trace.event('debug', 'integration', 'patch_accepted_after_gate_repair', {
|
||||
unitKey: input.unitKey,
|
||||
commitSha: commit.commitHash,
|
||||
touchedPaths: gateRepair.changedPaths,
|
||||
attempts: gateRepair.attempts,
|
||||
});
|
||||
return {
|
||||
status: 'accepted',
|
||||
commitSha: commit.commitHash,
|
||||
touchedPaths: gateRepair.changedPaths,
|
||||
gateRepair,
|
||||
};
|
||||
}
|
||||
|
||||
if (preApplyHead) {
|
||||
await input.integrationGit.resetHardTo(preApplyHead);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import type { IngestSessionWorktree, IngestSessionWorktreePort } from '../ports.
|
|||
import type { WorkUnit } from '../types.js';
|
||||
import type { IngestTraceWriter } from '../ingest-trace.js';
|
||||
import type { WorkUnitOutcome } from '../stages/stage-3-work-units.js';
|
||||
import { captureIngestWorkUnitCachedArtifactFiles } from '../work-unit-cache.js';
|
||||
import { parsePatchTouchedPaths } from './git-patch.js';
|
||||
|
||||
export interface RunIsolatedWorkUnitInput {
|
||||
|
|
@ -19,7 +20,7 @@ export interface RunIsolatedWorkUnitInput {
|
|||
afterSuccess?(child: IngestSessionWorktree): Promise<void>;
|
||||
}
|
||||
|
||||
function patchFileName(unitIndex: number, unitKey: string): string {
|
||||
export function workUnitPatchFileName(unitIndex: number, unitKey: string): string {
|
||||
const safeKey = unitKey.replace(/[^a-zA-Z0-9_.-]+/g, '-');
|
||||
return `${String(unitIndex).padStart(4, '0')}-${safeKey}.patch`;
|
||||
}
|
||||
|
|
@ -84,21 +85,29 @@ export async function runIsolatedWorkUnit(input: RunIsolatedWorkUnitInput): Prom
|
|||
|
||||
await input.afterSuccess?.(child);
|
||||
await mkdir(input.patchDir, { recursive: true });
|
||||
const patchPath = join(input.patchDir, patchFileName(input.unitIndex, input.workUnit.unitKey));
|
||||
const patchPath = join(input.patchDir, workUnitPatchFileName(input.unitIndex, input.workUnit.unitKey));
|
||||
await child.git.writeBinaryNoRenamePatch(input.ingestionBaseSha, 'HEAD', patchPath);
|
||||
const patch = await readFile(patchPath, 'utf-8');
|
||||
const touched = parsePatchTouchedPaths(patch);
|
||||
const patchTouchedPaths = touched.map((entry) => entry.path);
|
||||
const artifactFiles = await captureIngestWorkUnitCachedArtifactFiles({
|
||||
git: child.git,
|
||||
workdir: child.workdir,
|
||||
baseSha: input.ingestionBaseSha,
|
||||
patchTouchedPaths,
|
||||
});
|
||||
cleanupOutcome = 'success';
|
||||
await input.trace.event('debug', 'work_unit', 'work_unit_patch_collected', {
|
||||
unitKey: input.workUnit.unitKey,
|
||||
patchPath,
|
||||
touchedPaths: touched.map((entry) => entry.path),
|
||||
touchedPaths: patchTouchedPaths,
|
||||
patchBytes: Buffer.byteLength(patch),
|
||||
});
|
||||
return {
|
||||
...outcome,
|
||||
patchPath,
|
||||
patchTouchedPaths: touched.map((entry) => entry.path),
|
||||
patchTouchedPaths,
|
||||
artifactFiles,
|
||||
childWorktreePath: child.workdir,
|
||||
};
|
||||
} catch (error) {
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import { createLocalKtxLlmRuntimeFromConfig } from '../../context/llm/local-conf
|
|||
import { KtxIngestEmbeddingPortAdapter } from '../../context/llm/embedding-port.js';
|
||||
import { createRateLimitGovernorConfig, RateLimitGovernor } from '../../context/llm/rate-limit-governor.js';
|
||||
import { RuntimeAgentRunner, type AgentRunnerPort, type KtxLlmRuntimePort, type KtxRuntimeToolSet } from '../../context/llm/runtime-port.js';
|
||||
import { getKtxCliPackageInfo } from '../../cli-runtime.js';
|
||||
import type { KtxEmbeddingProvider } from '../../llm/types.js';
|
||||
import type { KtxLocalProject } from '../../context/project/project.js';
|
||||
import { ktxLocalStateDbPath } from '../../context/project/local-state-db.js';
|
||||
|
|
@ -54,6 +55,7 @@ import { WikiWriteTool } from '../../context/wiki/tools/wiki-write.tool.js';
|
|||
import { CandidateDedupService } from '../../context/ingest/context-candidates/candidate-dedup.service.js';
|
||||
import { ContextCandidateCarryforwardService } from '../../context/ingest/context-candidates/context-candidate-carryforward.service.js';
|
||||
import { CuratorPaginationService } from '../../context/ingest/context-candidates/curator-pagination.service.js';
|
||||
import { SqliteContentResultCache } from '../cache/sqlite-content-result-cache.js';
|
||||
import { createEmitHistoricSqlEvidenceTool } from './adapters/historic-sql/evidence-tool.js';
|
||||
import { ContextEvidenceIndexService } from '../../context/ingest/context-evidence/context-evidence-index.service.js';
|
||||
import { SqliteContextEvidenceStore } from '../../context/ingest/context-evidence/sqlite-context-evidence-store.js';
|
||||
|
|
@ -657,6 +659,7 @@ export function createLocalBundleIngestRuntime(
|
|||
mkdirSync(join(options.project.projectDir, '.ktx/cache/local-ingest'), { recursive: true });
|
||||
const store = new SqliteBundleIngestStore({ dbPath });
|
||||
const contextStore = new SqliteContextEvidenceStore({ dbPath });
|
||||
const contentCache = new SqliteContentResultCache({ dbPath });
|
||||
const embeddingProvider = options.embeddingProvider ?? null;
|
||||
if (!embeddingProvider && options.project.config.ingest.embeddings.backend !== 'none') {
|
||||
// Embedding-dependent stages (CandidateDedup clustering, ContextEvidenceIndex
|
||||
|
|
@ -711,6 +714,7 @@ export function createLocalBundleIngestRuntime(
|
|||
provenance: store,
|
||||
reports: store,
|
||||
canonicalPins: store,
|
||||
contentCache,
|
||||
registry,
|
||||
diffSetService: new DiffSetService(store),
|
||||
sessionWorktreeService: new SessionWorktreeService({
|
||||
|
|
@ -724,6 +728,7 @@ export function createLocalBundleIngestRuntime(
|
|||
storage,
|
||||
settings: {
|
||||
memoryIngestionModel: options.project.config.llm.models.default ?? 'local-ingest-model',
|
||||
cliVersion: getKtxCliPackageInfo().version,
|
||||
probeRowCount: 0,
|
||||
workUnitMaxConcurrency: options.project.config.ingest.workUnits.maxConcurrency,
|
||||
workUnitStepBudget: options.project.config.ingest.workUnits.stepBudget,
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import type { ToolContext } from '../../context/tools/base-tool.js';
|
|||
import type { ToolSession } from '../../context/tools/tool-session.js';
|
||||
import type { KnowledgeIndexPort } from '../../context/wiki/ports.js';
|
||||
import type { KnowledgeWikiService } from '../../context/wiki/knowledge-wiki.service.js';
|
||||
import type { ContentResultCache } from '../cache/content-result-cache.js';
|
||||
import type { CanonicalPin } from './canonical-pins.js';
|
||||
import type { IngestTraceLevel } from './ingest-trace.js';
|
||||
import type { IngestReportSnapshot } from './reports.js';
|
||||
|
|
@ -141,6 +142,7 @@ export interface IngestSessionWorktreePort {
|
|||
|
||||
interface IngestSettingsPort {
|
||||
memoryIngestionModel: string;
|
||||
cliVersion: string;
|
||||
probeRowCount: number;
|
||||
workUnitMaxConcurrency?: number;
|
||||
workUnitStepBudget?: number;
|
||||
|
|
@ -333,6 +335,7 @@ export interface IngestBundleRunnerDeps {
|
|||
provenance: IngestProvenancePort;
|
||||
reports: IngestReportsPort;
|
||||
canonicalPins: IngestCanonicalPinsPort;
|
||||
contentCache: ContentResultCache;
|
||||
registry: SourceAdapterRegistryPort;
|
||||
diffSetService: DiffSetComputerPort;
|
||||
sessionWorktreeService: IngestSessionWorktreePort;
|
||||
|
|
|
|||
|
|
@ -158,6 +158,19 @@ const finalizationOutcomeSchema = z.object({
|
|||
provenanceExclusions: z.array(finalizationProvenanceExclusionSchema).default([]),
|
||||
});
|
||||
|
||||
const finalGatePrunedReferenceSchema = z.object({
|
||||
kind: z.enum(['join', 'wiki_ref', 'wiki_sl_ref', 'wiki_body_ref']),
|
||||
artifact: z.string(),
|
||||
removedRef: z.string(),
|
||||
absentTarget: z.string(),
|
||||
});
|
||||
|
||||
const finalGateDroppedSourceSchema = z.object({
|
||||
connectionId: z.string(),
|
||||
sourceName: z.string(),
|
||||
reason: z.string(),
|
||||
});
|
||||
|
||||
const ingestReportSnapshotSchema = z
|
||||
.object({
|
||||
id: z.string().min(1),
|
||||
|
|
@ -187,9 +200,6 @@ const ingestReportSnapshotSchema = z
|
|||
resolverAttempts: z.number().int().min(0).default(0),
|
||||
resolverRepairs: z.number().int().min(0).default(0),
|
||||
resolverFailures: z.number().int().min(0).default(0),
|
||||
gateRepairAttempts: z.number().int().min(0).default(0),
|
||||
gateRepairs: z.number().int().min(0).default(0),
|
||||
gateRepairFailures: z.number().int().min(0).default(0),
|
||||
})
|
||||
.optional(),
|
||||
workUnits: z.array(
|
||||
|
|
@ -218,6 +228,8 @@ const ingestReportSnapshotSchema = z
|
|||
provenanceRows: z.array(provenanceDetailSchema).default([]),
|
||||
toolTranscripts: z.array(toolTranscriptSummarySchema).default([]),
|
||||
finalization: finalizationOutcomeSchema.optional(),
|
||||
finalGatePrunedReferences: z.array(finalGatePrunedReferenceSchema).default([]),
|
||||
finalGateDroppedSources: z.array(finalGateDroppedSourceSchema).default([]),
|
||||
memoryFlow: memoryFlowReplayInputSchema.optional(),
|
||||
})
|
||||
.passthrough(),
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import type { MemoryAction } from '../../context/memory/types.js';
|
||||
import type { TouchedSlSource } from '../../context/tools/touched-sl-sources.js';
|
||||
import type { MemoryFlowReplayInput } from './memory-flow/types.js';
|
||||
import type { FinalGateDroppedSource, FinalGatePrunedReference } from './final-gate-prune.js';
|
||||
import type { IngestProvenanceInsert } from './ports.js';
|
||||
import type {
|
||||
ArtifactResolutionRecord,
|
||||
|
|
@ -93,9 +94,6 @@ export interface IngestReportBody {
|
|||
resolverAttempts?: number;
|
||||
resolverRepairs?: number;
|
||||
resolverFailures?: number;
|
||||
gateRepairAttempts?: number;
|
||||
gateRepairs?: number;
|
||||
gateRepairFailures?: number;
|
||||
};
|
||||
workUnits: IngestReportWorkUnit[];
|
||||
failedWorkUnits: string[];
|
||||
|
|
@ -115,6 +113,8 @@ export interface IngestReportBody {
|
|||
provenanceRows: IngestReportProvenanceDetail[];
|
||||
toolTranscripts: IngestReportToolTranscriptSummary[];
|
||||
finalization?: IngestReportFinalizationOutcome;
|
||||
finalGatePrunedReferences?: FinalGatePrunedReference[];
|
||||
finalGateDroppedSources?: FinalGateDroppedSource[];
|
||||
wikiSlRefRepairs?: WikiSlRefRepair[];
|
||||
wikiSlRefRepairWarnings?: string[];
|
||||
memoryFlow?: MemoryFlowReplayInput;
|
||||
|
|
@ -153,7 +153,10 @@ export function ingestReportOutcome(report: IngestReportSnapshot): IngestReportO
|
|||
if (report.body.status === 'failed') {
|
||||
return 'error';
|
||||
}
|
||||
if (report.body.failedWorkUnits.length === 0) {
|
||||
const hasPruneOrDrop =
|
||||
(report.body.finalGatePrunedReferences?.length ?? 0) > 0 ||
|
||||
(report.body.finalGateDroppedSources?.length ?? 0) > 0;
|
||||
if (report.body.failedWorkUnits.length === 0 && !hasPruneOrDrop) {
|
||||
return 'done';
|
||||
}
|
||||
const { wikiCount, slCount } = savedMemoryCountsForReport(report);
|
||||
|
|
|
|||
|
|
@ -3,7 +3,12 @@ import { isAbortError } from '../../core/abort.js';
|
|||
import type { AgentRunnerPort, KtxRuntimeToolSet, RunLoopMetrics } from '../../../context/llm/runtime-port.js';
|
||||
import type { CaptureSession, MemoryAction } from '../../../context/memory/types.js';
|
||||
import { listTouchedSlSources, type TouchedSlSource } from '../../../context/tools/touched-sl-sources.js';
|
||||
import { formatInvalidWuSources, type WuValidationResult } from './validate-wu-sources.js';
|
||||
import {
|
||||
formatInvalidWuSources,
|
||||
hasBlockingWuSourceIssue,
|
||||
type WuValidationResult,
|
||||
} from './validate-wu-sources.js';
|
||||
import type { IngestWorkUnitCachedArtifactFile } from '../work-unit-cache.js';
|
||||
import type { WorkUnit } from '../types.js';
|
||||
|
||||
const MAX_WORK_UNIT_PROMPT_CHARS = 240_000;
|
||||
|
|
@ -11,7 +16,6 @@ const MAX_WORK_UNIT_PROMPT_CHARS = 240_000;
|
|||
export interface WorkUnitExecutionDeps {
|
||||
sessionWorktreeGit: { revParseHead(): Promise<string | null> };
|
||||
agentRunner: AgentRunnerPort;
|
||||
validateWikiRefs?: (actions: MemoryAction[]) => Promise<string[]>;
|
||||
validateTouchedSources: (touched: TouchedSlSource[]) => Promise<WuValidationResult>;
|
||||
resetHardTo: (targetSha: string) => Promise<void>;
|
||||
buildSystemPrompt: (wu: WorkUnit) => string;
|
||||
|
|
@ -40,6 +44,7 @@ export interface WorkUnitOutcome {
|
|||
slDisallowedReason?: 'lookml_connection_mismatch';
|
||||
patchPath?: string;
|
||||
patchTouchedPaths?: string[];
|
||||
artifactFiles?: IngestWorkUnitCachedArtifactFile[];
|
||||
childWorktreePath?: string;
|
||||
/** Timing and token metrics for the work-unit agent loop, used for ingest profiling. */
|
||||
metrics?: RunLoopMetrics;
|
||||
|
|
@ -140,19 +145,12 @@ export async function executeWorkUnit(deps: WorkUnitExecutionDeps, wu: WorkUnit)
|
|||
return failWithReset(`${toolFailureCount} tool call(s) failed during WorkUnit ${wu.unitKey}`);
|
||||
}
|
||||
|
||||
const danglingWikiRefs = (await deps.validateWikiRefs?.(deps.sessionActions)) ?? [];
|
||||
if (danglingWikiRefs.length > 0) {
|
||||
return failWithReset(`wiki references target missing page(s): ${danglingWikiRefs.join(', ')}`);
|
||||
}
|
||||
|
||||
const touched = listTouchedSlSources(deps.captureSession.touchedSlSources);
|
||||
if (touched.length > 0) {
|
||||
const validation = await deps.validateTouchedSources(touched);
|
||||
if (validation.invalidSources.length > 0) {
|
||||
// Spec: invalid SL writes reset the session worktree to the WU's pre-state, WU is marked failed,
|
||||
// its files are absent from the Stage Index. Per-source surgical revert is the
|
||||
// memory-agent pattern — NOT the bundle-ingest pattern.
|
||||
return failWithReset(`sl_validate failed for: ${formatInvalidWuSources(validation.invalidSources)}`);
|
||||
const blockingInvalidSources = validation.invalidSources.filter(hasBlockingWuSourceIssue);
|
||||
if (blockingInvalidSources.length > 0) {
|
||||
return failWithReset(`sl_validate failed for: ${formatInvalidWuSources(blockingInvalidSources)}`);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ export interface InvalidWuSource {
|
|||
/** `${connectionId}:${sourceName}` */
|
||||
source: string;
|
||||
errors: string[];
|
||||
issues?: WuValidationIssue[];
|
||||
}
|
||||
|
||||
export interface WuValidationResult {
|
||||
|
|
@ -14,10 +15,24 @@ export interface WuValidationResult {
|
|||
invalidSources: InvalidWuSource[];
|
||||
}
|
||||
|
||||
type WuValidationIssue =
|
||||
| { kind: 'source_validation'; message: string }
|
||||
| { kind: 'missing_join_target'; targetSourceName: string; caseMismatch: string | null; message: string };
|
||||
|
||||
export function formatInvalidWuSources(invalid: InvalidWuSource[]): string {
|
||||
return invalid.map((entry) => `${entry.source} (${entry.errors.join('; ')})`).join(', ');
|
||||
}
|
||||
|
||||
export function hasBlockingWuSourceIssue(source: InvalidWuSource): boolean {
|
||||
const issues =
|
||||
source.issues ??
|
||||
source.errors.map((message) => ({
|
||||
kind: 'source_validation' as const,
|
||||
message,
|
||||
}));
|
||||
return issues.some((issue) => issue.kind === 'source_validation');
|
||||
}
|
||||
|
||||
type LoadedSource = Awaited<ReturnType<SlValidationDeps['semanticLayerService']['loadAllSources']>>['sources'][number];
|
||||
|
||||
function uniqueTouchedSources(sources: TouchedSlSource[]): TouchedSlSource[] {
|
||||
|
|
@ -86,11 +101,11 @@ function expandWithExistingJoinNeighbors(
|
|||
* are out of scope — they must not block unrelated work. Resolution is the
|
||||
* Python engine's: exact source-name match within the connection.
|
||||
*/
|
||||
function findJoinTargetErrors(
|
||||
function findJoinTargetIssues(
|
||||
touched: TouchedSlSource[],
|
||||
sourcesByConnection: Map<string, LoadedSource[]>,
|
||||
): Map<string, string[]> {
|
||||
const errorsBySource = new Map<string, string[]>();
|
||||
): Map<string, WuValidationIssue[]> {
|
||||
const issuesBySource = new Map<string, WuValidationIssue[]>();
|
||||
const touchedByConnection = new Map<string, Set<string>>();
|
||||
for (const source of touched) {
|
||||
const bucket = touchedByConnection.get(source.connectionId) ?? new Set<string>();
|
||||
|
|
@ -114,11 +129,16 @@ function findJoinTargetErrors(
|
|||
continue;
|
||||
}
|
||||
const key = `${connectionId}:${source.name}`;
|
||||
const messages = missing.map(formatMissingJoinTarget);
|
||||
errorsBySource.set(key, [...(errorsBySource.get(key) ?? []), ...messages]);
|
||||
const issues = missing.map((entry) => ({
|
||||
kind: 'missing_join_target' as const,
|
||||
targetSourceName: entry.to,
|
||||
caseMismatch: entry.caseMismatch,
|
||||
message: formatMissingJoinTarget(entry),
|
||||
}));
|
||||
issuesBySource.set(key, [...(issuesBySource.get(key) ?? []), ...issues]);
|
||||
}
|
||||
}
|
||||
return errorsBySource;
|
||||
return issuesBySource;
|
||||
}
|
||||
|
||||
export async function validateWuTouchedSources(
|
||||
|
|
@ -136,18 +156,20 @@ export async function validateWuTouchedSources(
|
|||
}
|
||||
|
||||
const expanded = expandWithExistingJoinNeighbors(touched, sourcesByConnection);
|
||||
const joinTargetErrors = findJoinTargetErrors(touched, sourcesByConnection);
|
||||
const joinTargetIssues = findJoinTargetIssues(touched, sourcesByConnection);
|
||||
|
||||
const valid: string[] = [];
|
||||
const invalid: InvalidWuSource[] = [];
|
||||
for (const source of expanded) {
|
||||
const key = `${source.connectionId}:${source.sourceName}`;
|
||||
const result = await deps.slValidator.validateSingleSource(deps, source.connectionId, source.sourceName);
|
||||
const errors = [...result.errors, ...(joinTargetErrors.get(key) ?? [])];
|
||||
const sourceIssues: WuValidationIssue[] = result.errors.map((message) => ({ kind: 'source_validation', message }));
|
||||
const issues = [...sourceIssues, ...(joinTargetIssues.get(key) ?? [])];
|
||||
const errors = issues.map((issue) => issue.message);
|
||||
if (errors.length === 0) {
|
||||
valid.push(key);
|
||||
} else {
|
||||
invalid.push({ source: key, errors });
|
||||
invalid.push({ source: key, errors, issues });
|
||||
}
|
||||
}
|
||||
return { validSources: valid, invalidSources: invalid };
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import type { KtxTableRefKey } from '../scan/table-ref.js';
|
|||
import type { MemoryFlowEventSink } from './memory-flow/types.js';
|
||||
import type { StageIndex } from './stages/stage-index.types.js';
|
||||
import type { WorkUnitOutcome } from './stages/stage-3-work-units.js';
|
||||
import type { FinalGateDroppedSource, FinalGatePrunedReference } from './final-gate-prune.js';
|
||||
|
||||
export type IngestTrigger = 'upload' | 'scheduled_pull' | 'manual_resync' | 'manual_override';
|
||||
|
||||
|
|
@ -210,6 +211,8 @@ export interface IngestBundleResult {
|
|||
failedWorkUnits: string[];
|
||||
artifactsWritten: number;
|
||||
commitSha: string | null;
|
||||
finalGatePrunedReferences?: FinalGatePrunedReference[];
|
||||
finalGateDroppedSources?: FinalGateDroppedSource[];
|
||||
}
|
||||
|
||||
export interface IngestJobPhase {
|
||||
|
|
|
|||
|
|
@ -2,9 +2,9 @@ import type { SemanticLayerSource } from '../../context/sl/types.js';
|
|||
|
||||
/** @internal */
|
||||
export type WikiBodyRef =
|
||||
| { kind: 'sl_entity'; connectionId: string | null; sourceName: string; entityName: string }
|
||||
| { kind: 'sl_source'; connectionId: string | null; sourceName: string }
|
||||
| { kind: 'table'; connectionId: string | null; tableRef: string };
|
||||
| { kind: 'sl_entity'; connectionId: string | null; sourceName: string; entityName: string; rawToken: string }
|
||||
| { kind: 'sl_source'; connectionId: string | null; sourceName: string; rawToken: string }
|
||||
| { kind: 'table'; connectionId: string | null; tableRef: string; rawToken: string };
|
||||
|
||||
export interface WikiBodyRefValidationInput {
|
||||
pageKey: string;
|
||||
|
|
@ -14,6 +14,33 @@ export interface WikiBodyRefValidationInput {
|
|||
tableExists(connectionId: string, tableRef: string): Promise<boolean>;
|
||||
}
|
||||
|
||||
export type WikiBodyRefIssue =
|
||||
| {
|
||||
kind: 'missing_wiki_body_sl_entity';
|
||||
pageKey: string;
|
||||
rawToken: string;
|
||||
connectionId?: string;
|
||||
sourceName: string;
|
||||
entityName: string;
|
||||
message: string;
|
||||
}
|
||||
| {
|
||||
kind: 'missing_wiki_body_sl_source';
|
||||
pageKey: string;
|
||||
rawToken: string;
|
||||
connectionId?: string;
|
||||
sourceName: string;
|
||||
message: string;
|
||||
}
|
||||
| {
|
||||
kind: 'missing_wiki_body_table';
|
||||
pageKey: string;
|
||||
rawToken: string;
|
||||
connectionId?: string;
|
||||
tableRef: string;
|
||||
message: string;
|
||||
};
|
||||
|
||||
const inlineCodePattern = /`([^`\n]+)`/g;
|
||||
|
||||
function visibleLinesOutsideFences(body: string): string[] {
|
||||
|
|
@ -56,14 +83,14 @@ export function parseWikiBodyRefs(body: string): WikiBodyRef[] {
|
|||
if (scoped.body.startsWith('source:')) {
|
||||
const sourceName = scoped.body.slice('source:'.length).trim();
|
||||
if (sourceName) {
|
||||
refs.push({ kind: 'sl_source', connectionId: scoped.connectionId, sourceName });
|
||||
refs.push({ kind: 'sl_source', connectionId: scoped.connectionId, sourceName, rawToken: token });
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (scoped.body.startsWith('table:')) {
|
||||
const tableRef = scoped.body.slice('table:'.length).trim();
|
||||
if (tableRef) {
|
||||
refs.push({ kind: 'table', connectionId: scoped.connectionId, tableRef });
|
||||
refs.push({ kind: 'table', connectionId: scoped.connectionId, tableRef, rawToken: token });
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
|
@ -74,6 +101,7 @@ export function parseWikiBodyRefs(body: string): WikiBodyRef[] {
|
|||
connectionId: scoped.connectionId,
|
||||
sourceName: parts[0],
|
||||
entityName: parts[1],
|
||||
rawToken: token,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -89,8 +117,8 @@ function entityNames(source: SemanticLayerSource): Set<string> {
|
|||
]);
|
||||
}
|
||||
|
||||
export async function findInvalidWikiBodyRefs(input: WikiBodyRefValidationInput): Promise<string[]> {
|
||||
const errors: string[] = [];
|
||||
export async function findInvalidWikiBodyRefIssues(input: WikiBodyRefValidationInput): Promise<WikiBodyRefIssue[]> {
|
||||
const issues: WikiBodyRefIssue[] = [];
|
||||
const sourceCache = new Map<string, SemanticLayerSource[]>();
|
||||
const loadSources = async (connectionId: string): Promise<SemanticLayerSource[]> => {
|
||||
const cached = sourceCache.get(connectionId);
|
||||
|
|
@ -120,7 +148,15 @@ export async function findInvalidWikiBodyRefs(input: WikiBodyRefValidationInput)
|
|||
if (ref.kind === 'table') {
|
||||
const found = await Promise.all(connectionIds.map((connectionId) => input.tableExists(connectionId, ref.tableRef)));
|
||||
if (!found.some(Boolean)) {
|
||||
errors.push(`${input.pageKey}: unknown raw table ${ref.connectionId ? `${ref.connectionId}/` : ''}${ref.tableRef}`);
|
||||
const target = `${ref.connectionId ? `${ref.connectionId}/` : ''}${ref.tableRef}`;
|
||||
issues.push({
|
||||
kind: 'missing_wiki_body_table',
|
||||
pageKey: input.pageKey,
|
||||
rawToken: ref.rawToken,
|
||||
...(ref.connectionId ? { connectionId: ref.connectionId } : {}),
|
||||
tableRef: ref.tableRef,
|
||||
message: `${input.pageKey}: unknown raw table ${target}`,
|
||||
});
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
|
@ -128,16 +164,35 @@ export async function findInvalidWikiBodyRefs(input: WikiBodyRefValidationInput)
|
|||
const found = await findSource(connectionIds, ref.sourceName);
|
||||
if (!found) {
|
||||
if (ref.kind === 'sl_source') {
|
||||
errors.push(
|
||||
`${input.pageKey}: unknown semantic-layer source ${ref.connectionId ? `${ref.connectionId}/` : ''}${ref.sourceName}`,
|
||||
);
|
||||
const target = `${ref.connectionId ? `${ref.connectionId}/` : ''}${ref.sourceName}`;
|
||||
issues.push({
|
||||
kind: 'missing_wiki_body_sl_source',
|
||||
pageKey: input.pageKey,
|
||||
rawToken: ref.rawToken,
|
||||
...(ref.connectionId ? { connectionId: ref.connectionId } : {}),
|
||||
sourceName: ref.sourceName,
|
||||
message: `${input.pageKey}: unknown semantic-layer source ${target}`,
|
||||
});
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (ref.kind === 'sl_entity' && !entityNames(found.source).has(ref.entityName)) {
|
||||
errors.push(`${input.pageKey}: unknown semantic-layer entity ${ref.sourceName}.${ref.entityName}`);
|
||||
issues.push({
|
||||
kind: 'missing_wiki_body_sl_entity',
|
||||
pageKey: input.pageKey,
|
||||
rawToken: ref.rawToken,
|
||||
...(ref.connectionId ? { connectionId: ref.connectionId } : {}),
|
||||
sourceName: ref.sourceName,
|
||||
entityName: ref.entityName,
|
||||
message: `${input.pageKey}: unknown semantic-layer entity ${ref.sourceName}.${ref.entityName}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return errors;
|
||||
return issues;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export async function findInvalidWikiBodyRefs(input: WikiBodyRefValidationInput): Promise<string[]> {
|
||||
return (await findInvalidWikiBodyRefIssues(input)).map((issue) => issue.message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -82,6 +82,7 @@ export async function repairWikiSlRefs(input: {
|
|||
semanticLayerService: SemanticLayerService;
|
||||
configService: KtxFileStorePort;
|
||||
connectionIds: string[];
|
||||
deferGlobalPageKeys?: string[];
|
||||
}): Promise<WikiSlRefRepairResult> {
|
||||
const { refs: validRefs, warnings } = await loadVisibleSlRefs(input.semanticLayerService, input.connectionIds);
|
||||
const listFiles =
|
||||
|
|
@ -96,12 +97,16 @@ export async function repairWikiSlRefs(input: {
|
|||
}
|
||||
const listed = await listFiles('wiki', true);
|
||||
const repairs: WikiSlRefRepair[] = [];
|
||||
const deferredGlobalPageKeys = new Set(input.deferGlobalPageKeys ?? []);
|
||||
|
||||
for (const file of listed.files.sort()) {
|
||||
const parsedPath = parseKnowledgeFilePath(file);
|
||||
if (!parsedPath) {
|
||||
continue;
|
||||
}
|
||||
if (parsedPath.scope === 'GLOBAL' && deferredGlobalPageKeys.has(parsedPath.pageKey)) {
|
||||
continue;
|
||||
}
|
||||
const page = await input.wikiService.readPage(parsedPath.scope, parsedPath.scopeId, parsedPath.pageKey);
|
||||
const refs = uniqueStringArray(page?.frontmatter.sl_refs);
|
||||
if (!page || refs.length === 0) {
|
||||
|
|
|
|||
314
packages/cli/src/context/ingest/work-unit-cache.ts
Normal file
314
packages/cli/src/context/ingest/work-unit-cache.ts
Normal file
|
|
@ -0,0 +1,314 @@
|
|||
import { mkdir, readFile, rm, writeFile } from 'node:fs/promises';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { isDeepStrictEqual } from 'node:util';
|
||||
import YAML from 'yaml';
|
||||
import type { KtxModelRole } from '../../llm/types.js';
|
||||
import { stableContentHash } from '../cache/content-result-cache.js';
|
||||
import type { GitService } from '../core/git.service.js';
|
||||
import type { MemoryAction } from '../memory/types.js';
|
||||
import type { TouchedSlSource } from '../tools/touched-sl-sources.js';
|
||||
import type { IngestTraceWriter } from './ingest-trace.js';
|
||||
import type { IngestSessionWorktreePort } from './ports.js';
|
||||
import type { WorkUnit } from './types.js';
|
||||
|
||||
export const INGEST_WORK_UNIT_CACHE_NAMESPACE = 'ingest:work-unit';
|
||||
|
||||
export interface IngestWorkUnitCachedArtifactFile {
|
||||
path: string;
|
||||
beforeBase64: string | null;
|
||||
afterBase64: string | null;
|
||||
}
|
||||
|
||||
export interface IngestWorkUnitCachePayload {
|
||||
schemaVersion: 2;
|
||||
unitKey: string;
|
||||
patchTouchedPaths: string[];
|
||||
// Replay re-derives the patch from these before/after snapshots; the diff text
|
||||
// itself is never stored, so the payload carries each touched file only once.
|
||||
artifactFiles: IngestWorkUnitCachedArtifactFile[];
|
||||
actions: MemoryAction[];
|
||||
touchedSlSources: TouchedSlSource[];
|
||||
slDisallowed?: boolean;
|
||||
slDisallowedReason?: 'lookml_connection_mismatch';
|
||||
}
|
||||
|
||||
export interface ComputeIngestWorkUnitInputHashInput {
|
||||
stagedDir: string;
|
||||
connectionId: string;
|
||||
sourceKey: string;
|
||||
unit: WorkUnit;
|
||||
cliVersion: string;
|
||||
promptFingerprint: string;
|
||||
modelRole: KtxModelRole;
|
||||
}
|
||||
|
||||
async function fileDigest(
|
||||
stagedDir: string,
|
||||
path: string,
|
||||
): Promise<{ path: string; status: 'present' | 'missing'; hash: string | null }> {
|
||||
try {
|
||||
const bytes = await readFile(join(stagedDir, path));
|
||||
return { path, status: 'present', hash: stableContentHash(bytes.toString('base64')) };
|
||||
} catch (error) {
|
||||
if (error && typeof error === 'object' && 'code' in error && error.code === 'ENOENT') {
|
||||
return { path, status: 'missing', hash: null };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function computeIngestWorkUnitInputHash(input: ComputeIngestWorkUnitInputHashInput): Promise<string> {
|
||||
const rawFiles = [...input.unit.rawFiles].sort();
|
||||
const dependencyPaths = [...input.unit.dependencyPaths].sort();
|
||||
const [raw, dependencies] = await Promise.all([
|
||||
Promise.all(rawFiles.map((path) => fileDigest(input.stagedDir, path))),
|
||||
Promise.all(dependencyPaths.map((path) => fileDigest(input.stagedDir, path))),
|
||||
]);
|
||||
|
||||
return stableContentHash({
|
||||
schemaVersion: 2,
|
||||
connectionId: input.connectionId,
|
||||
sourceKey: input.sourceKey,
|
||||
unitKey: input.unit.unitKey,
|
||||
rawFiles: raw,
|
||||
dependencyPaths: dependencies,
|
||||
slDisallowed: input.unit.slDisallowed === true,
|
||||
slDisallowedReason: input.unit.slDisallowedReason ?? null,
|
||||
cliVersion: input.cliVersion,
|
||||
promptFingerprint: input.promptFingerprint,
|
||||
modelRole: input.modelRole,
|
||||
});
|
||||
}
|
||||
|
||||
async function readFileBase64(workdir: string, path: string): Promise<string | null> {
|
||||
try {
|
||||
return (await readFile(join(workdir, path))).toString('base64');
|
||||
} catch (error) {
|
||||
if (error && typeof error === 'object' && 'code' in error && error.code === 'ENOENT') {
|
||||
return null;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function readGitFileBase64(git: GitService, path: string, commitSha: string): Promise<string | null> {
|
||||
try {
|
||||
return Buffer.from(await git.getFileAtCommit(path, commitSha), 'utf-8').toString('base64');
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function decodeBase64(value: string | null): string | null {
|
||||
return value === null ? null : Buffer.from(value, 'base64').toString('utf-8');
|
||||
}
|
||||
|
||||
function parseYamlObject(content: string): Record<string, unknown> | null {
|
||||
const parsed = YAML.parse(content);
|
||||
return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as Record<string, unknown>) : null;
|
||||
}
|
||||
|
||||
function isSubsequenceOfDeepValues(current: unknown[], output: unknown[]): boolean {
|
||||
let outputIndex = 0;
|
||||
for (const item of current) {
|
||||
while (outputIndex < output.length && !isDeepStrictEqual(item, output[outputIndex])) {
|
||||
outputIndex += 1;
|
||||
}
|
||||
if (outputIndex >= output.length) {
|
||||
return false;
|
||||
}
|
||||
outputIndex += 1;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function isSemanticLayerPruneShape(current: string, output: string): boolean {
|
||||
const currentYaml = parseYamlObject(current);
|
||||
const outputYaml = parseYamlObject(output);
|
||||
if (!currentYaml || !outputYaml) {
|
||||
return false;
|
||||
}
|
||||
const currentJoins = Array.isArray(currentYaml.joins) ? currentYaml.joins : [];
|
||||
const outputJoins = Array.isArray(outputYaml.joins) ? outputYaml.joins : [];
|
||||
if (currentJoins.length >= outputJoins.length) {
|
||||
return false;
|
||||
}
|
||||
if (!isSubsequenceOfDeepValues(currentJoins, outputJoins)) {
|
||||
return false;
|
||||
}
|
||||
const normalizedOutput = { ...outputYaml, joins: currentJoins };
|
||||
return isDeepStrictEqual(currentYaml, normalizedOutput);
|
||||
}
|
||||
|
||||
function parseWikiPage(raw: string): { frontmatter: Record<string, unknown>; content: string } | null {
|
||||
const match = raw.match(/^---\n([\s\S]*?)\n---\n?([\s\S]*)$/);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const frontmatter = YAML.parse(match[1] ?? '') as Record<string, unknown>;
|
||||
return { frontmatter, content: (match[2] ?? '').trim() };
|
||||
}
|
||||
|
||||
function withoutRemovedWikiTokens(output: string, current: string): string {
|
||||
let projected = output;
|
||||
for (const match of output.matchAll(/\[\[\s*([^|\]\n]+)(?:\|[^\]\n]+)?\s*\]\]/g)) {
|
||||
const token = match[0] ?? '';
|
||||
if (token && !current.includes(token)) {
|
||||
projected = projected.replaceAll(token, '').replace(/[ \t]+([.,;:!?])/g, '$1');
|
||||
}
|
||||
}
|
||||
for (const match of output.matchAll(/`([^`\n]+)`/g)) {
|
||||
const token = match[0] ?? '';
|
||||
if (token && !current.includes(token)) {
|
||||
projected = projected.replaceAll(token, '').replace(/[ \t]+([.,;:!?])/g, '$1');
|
||||
}
|
||||
}
|
||||
projected = projected.replace(/,\s*,/g, ',').replace(/[ \t]+([.,;:!?])/g, '$1');
|
||||
return projected.trim();
|
||||
}
|
||||
|
||||
function isWikiPruneShape(current: string, output: string): boolean {
|
||||
const currentPage = parseWikiPage(current);
|
||||
const outputPage = parseWikiPage(output);
|
||||
if (!currentPage || !outputPage) {
|
||||
return false;
|
||||
}
|
||||
const currentRefs = Array.isArray(currentPage.frontmatter.refs) ? currentPage.frontmatter.refs : [];
|
||||
const outputRefs = Array.isArray(outputPage.frontmatter.refs) ? outputPage.frontmatter.refs : [];
|
||||
const currentSlRefs = Array.isArray(currentPage.frontmatter.sl_refs) ? currentPage.frontmatter.sl_refs : [];
|
||||
const outputSlRefs = Array.isArray(outputPage.frontmatter.sl_refs) ? outputPage.frontmatter.sl_refs : [];
|
||||
if (currentRefs.length > outputRefs.length || currentSlRefs.length > outputSlRefs.length) {
|
||||
return false;
|
||||
}
|
||||
if (!isSubsequenceOfDeepValues(currentRefs, outputRefs) || !isSubsequenceOfDeepValues(currentSlRefs, outputSlRefs)) {
|
||||
return false;
|
||||
}
|
||||
const normalizedOutputFrontmatter = {
|
||||
...outputPage.frontmatter,
|
||||
refs: currentRefs,
|
||||
sl_refs: currentSlRefs,
|
||||
};
|
||||
if (!isDeepStrictEqual(currentPage.frontmatter, normalizedOutputFrontmatter)) {
|
||||
return false;
|
||||
}
|
||||
return withoutRemovedWikiTokens(outputPage.content, currentPage.content) === currentPage.content.trim();
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export function isPruneShapedCachedReplayBase(path: string, currentContent: string, outputContent: string): boolean {
|
||||
if (path.startsWith('semantic-layer/') && path.endsWith('.yaml')) {
|
||||
return isSemanticLayerPruneShape(currentContent, outputContent);
|
||||
}
|
||||
if (path.startsWith('wiki/') && path.endsWith('.md')) {
|
||||
return isWikiPruneShape(currentContent, outputContent);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function captureIngestWorkUnitCachedArtifactFiles(input: {
|
||||
git: GitService;
|
||||
workdir: string;
|
||||
baseSha: string;
|
||||
patchTouchedPaths: string[];
|
||||
}): Promise<IngestWorkUnitCachedArtifactFile[]> {
|
||||
const paths = [...new Set(input.patchTouchedPaths)].sort();
|
||||
return Promise.all(
|
||||
paths.map(async (path) => ({
|
||||
path,
|
||||
beforeBase64: await readGitFileBase64(input.git, path, input.baseSha),
|
||||
afterBase64: await readFileBase64(input.workdir, path),
|
||||
})),
|
||||
);
|
||||
}
|
||||
|
||||
async function writeCachedFile(workdir: string, file: IngestWorkUnitCachedArtifactFile): Promise<void> {
|
||||
const target = join(workdir, file.path);
|
||||
if (file.afterBase64 === null) {
|
||||
await rm(target, { force: true });
|
||||
return;
|
||||
}
|
||||
await mkdir(dirname(target), { recursive: true });
|
||||
await writeFile(target, Buffer.from(file.afterBase64, 'base64'));
|
||||
}
|
||||
|
||||
function cacheFileCanReplayFromCurrentBase(file: IngestWorkUnitCachedArtifactFile, currentBase64: string | null): boolean {
|
||||
if (currentBase64 === file.beforeBase64 || currentBase64 === file.afterBase64) {
|
||||
return true;
|
||||
}
|
||||
const current = decodeBase64(currentBase64);
|
||||
const output = decodeBase64(file.afterBase64);
|
||||
if (current === null || output === null) {
|
||||
return false;
|
||||
}
|
||||
return isPruneShapedCachedReplayBase(file.path, current, output);
|
||||
}
|
||||
|
||||
export async function materializeCachedWorkUnitReplayPatch(input: {
|
||||
sessionWorktreeService: IngestSessionWorktreePort;
|
||||
baseSha: string;
|
||||
jobId: string;
|
||||
unitKey: string;
|
||||
patchPath: string;
|
||||
artifactFiles: IngestWorkUnitCachedArtifactFile[];
|
||||
author: { name: string; email: string };
|
||||
trace: IngestTraceWriter;
|
||||
}): Promise<'materialized' | 'unsafe_drift'> {
|
||||
const replay = await input.sessionWorktreeService.create(`${input.jobId}-${input.unitKey}-cache-replay`, input.baseSha);
|
||||
let cleanup: 'success' | 'crash' = 'crash';
|
||||
try {
|
||||
for (const file of input.artifactFiles) {
|
||||
const currentBase64 = await readFileBase64(replay.workdir, file.path);
|
||||
if (!cacheFileCanReplayFromCurrentBase(file, currentBase64)) {
|
||||
cleanup = 'success';
|
||||
return 'unsafe_drift';
|
||||
}
|
||||
}
|
||||
for (const file of input.artifactFiles) {
|
||||
await writeCachedFile(replay.workdir, file);
|
||||
}
|
||||
const changedPaths = await replay.git.changedPaths();
|
||||
if (changedPaths.length > 0) {
|
||||
await replay.git.commitFiles(
|
||||
changedPaths,
|
||||
`ingest: materialize cached WorkUnit ${input.unitKey}`,
|
||||
input.author.name,
|
||||
input.author.email,
|
||||
);
|
||||
}
|
||||
await replay.git.writeBinaryNoRenamePatch(input.baseSha, 'HEAD', input.patchPath);
|
||||
await input.trace.event('debug', 'work_unit', 'work_unit_cache_patch_materialized', {
|
||||
unitKey: input.unitKey,
|
||||
patchPath: input.patchPath,
|
||||
touchedPaths: changedPaths,
|
||||
});
|
||||
cleanup = 'success';
|
||||
return 'materialized';
|
||||
} finally {
|
||||
await input.sessionWorktreeService.cleanup(replay, cleanup);
|
||||
}
|
||||
}
|
||||
|
||||
export function ingestWorkUnitCacheScopeKey(input: { connectionId: string; sourceKey: string }): string {
|
||||
return `${input.connectionId}:${input.sourceKey}`;
|
||||
}
|
||||
|
||||
export function computeIngestWorkUnitPromptFingerprint(input: {
|
||||
cliVersion: string;
|
||||
baseFraming: string;
|
||||
skillsPrompt: string;
|
||||
canonicalPins: unknown[];
|
||||
sourceKey: string;
|
||||
connectionId: string;
|
||||
skillNames: string[];
|
||||
}): string {
|
||||
return stableContentHash({
|
||||
schemaVersion: 1,
|
||||
cliVersion: input.cliVersion,
|
||||
baseFraming: input.baseFraming,
|
||||
skillsPrompt: input.skillsPrompt,
|
||||
canonicalPins: input.canonicalPins,
|
||||
sourceKey: input.sourceKey,
|
||||
connectionId: input.connectionId,
|
||||
skillNames: [...input.skillNames].sort(),
|
||||
});
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
import { createHash } from 'node:crypto';
|
||||
import { stableContentHash } from '../cache/content-result-cache.js';
|
||||
import type { KtxScanRelationshipConfig } from '../project/config.js';
|
||||
import type { KtxScanEnrichmentStage, KtxScanEnrichmentStateSummary, KtxScanMode, KtxSchemaSnapshot } from './types.js';
|
||||
|
||||
|
|
@ -99,29 +99,12 @@ export interface KtxRelationshipsStageHashInput {
|
|||
llmIdentity: KtxScanLlmIdentity;
|
||||
}
|
||||
|
||||
function stableJson(value: unknown): string {
|
||||
if (Array.isArray(value)) {
|
||||
return `[${value.map(stableJson).join(',')}]`;
|
||||
}
|
||||
if (value && typeof value === 'object') {
|
||||
const entries = Object.entries(value as Record<string, unknown>).sort(([left], [right]) =>
|
||||
left.localeCompare(right),
|
||||
);
|
||||
return `{${entries.map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(',')}}`;
|
||||
}
|
||||
return JSON.stringify(value);
|
||||
}
|
||||
|
||||
function sha256(value: unknown): string {
|
||||
return createHash('sha256').update(stableJson(value)).digest('hex');
|
||||
}
|
||||
|
||||
export function computeKtxDescriptionsStageHash(input: KtxDescriptionsStageHashInput): string {
|
||||
return sha256({ snapshot: input.snapshot, llmIdentity: input.llmIdentity });
|
||||
return stableContentHash({ snapshot: input.snapshot, llmIdentity: input.llmIdentity });
|
||||
}
|
||||
|
||||
export function computeKtxEmbeddingsStageHash(input: KtxEmbeddingsStageHashInput): string {
|
||||
return sha256({
|
||||
return stableContentHash({
|
||||
snapshot: input.snapshot,
|
||||
embeddingIdentity: input.embeddingIdentity,
|
||||
descriptionDigest: input.descriptionDigest,
|
||||
|
|
@ -129,7 +112,7 @@ export function computeKtxEmbeddingsStageHash(input: KtxEmbeddingsStageHashInput
|
|||
}
|
||||
|
||||
export function computeKtxRelationshipsStageHash(input: KtxRelationshipsStageHashInput): string {
|
||||
return sha256({
|
||||
return stableContentHash({
|
||||
snapshot: input.snapshot,
|
||||
relationshipSettings: input.relationshipSettings,
|
||||
llmIdentity: input.llmIdentity,
|
||||
|
|
@ -143,7 +126,7 @@ export function computeKtxRelationshipsStageHash(input: KtxRelationshipsStageHas
|
|||
* that depend on the changed text (D4 self-healing).
|
||||
*/
|
||||
export function computeKtxScanDescriptionDigest(texts: readonly string[]): string {
|
||||
return sha256(texts);
|
||||
return stableContentHash(texts);
|
||||
}
|
||||
|
||||
function uniqueStages(stages: KtxScanEnrichmentStage[]): KtxScanEnrichmentStage[] {
|
||||
|
|
|
|||
|
|
@ -809,17 +809,21 @@ export async function runLocalScanEnrichment(
|
|||
let relationshipPartial: { reason: KtxRelationshipDetectionStopReason } | null = null;
|
||||
let relationships: KtxScanRelationshipSummary = { accepted: 0, review: 0, rejected: 0, skipped: 0 };
|
||||
|
||||
// Promote the paid descriptions + embeddings to the queryable layer at the
|
||||
// cost boundary, before the slow, kill-prone relationship stage — so an
|
||||
// interrupted relationship stage degrades to "no joins," never "no descriptions."
|
||||
if (shouldDetectRelationships && summary.tableDescriptions === 'completed' && input.onCheckpoint) {
|
||||
// Promote any non-relationship stage that ran this invocation (descriptions or
|
||||
// a `--stages embeddings,relationships` re-embed) before the slow, kill-prone
|
||||
// relationship stage, so an interruption degrades to "no joins," never lost
|
||||
// enrichment. descriptionUpdates uses the best-available set (D3): the manifest
|
||||
// merge overwrites scan-managed descriptions, so the empty this-invocation set
|
||||
// would delete prior on-disk ones.
|
||||
const checkpointablePaidWork = summary.tableDescriptions === 'completed' || summary.embeddings === 'completed';
|
||||
if (shouldDetectRelationships && checkpointablePaidWork && input.onCheckpoint) {
|
||||
await input.onCheckpoint({
|
||||
snapshot,
|
||||
summary: { ...summary },
|
||||
relationships,
|
||||
state: summarizeKtxScanEnrichmentState(state),
|
||||
warnings: [...warnings],
|
||||
descriptionUpdates: descriptions,
|
||||
descriptionUpdates: await resolveDownstreamDescriptions(),
|
||||
embeddingUpdates,
|
||||
relationshipUpdate: null,
|
||||
relationshipProfile: null,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import { mkdirSync } from 'node:fs';
|
||||
import { dirname } from 'node:path';
|
||||
import Database from 'better-sqlite3';
|
||||
import type { ContentResultCache, ContentResultCacheRecord } from '../cache/content-result-cache.js';
|
||||
import { SqliteContentResultCache } from '../cache/sqlite-content-result-cache.js';
|
||||
import type {
|
||||
KtxScanEnrichmentCompletedStage,
|
||||
KtxScanEnrichmentFailedStage,
|
||||
|
|
@ -8,281 +7,142 @@ import type {
|
|||
KtxScanEnrichmentStageRecord,
|
||||
KtxScanEnrichmentStateStore,
|
||||
} from './enrichment-state.js';
|
||||
import { KTX_SCAN_ENRICHMENT_STAGES } from './enrichment-state.js';
|
||||
import { KTX_SCAN_MODES } from './types.js';
|
||||
import type { KtxScanEnrichmentStage, KtxScanMode } from './types.js';
|
||||
|
||||
export interface SqliteLocalScanEnrichmentStateStoreOptions {
|
||||
dbPath: string;
|
||||
cache?: ContentResultCache;
|
||||
}
|
||||
|
||||
interface StageRow {
|
||||
run_id: string;
|
||||
connection_id: string;
|
||||
sync_id: string;
|
||||
interface ScanStageMetadata {
|
||||
connectionId: string;
|
||||
syncId: string;
|
||||
mode: KtxScanMode;
|
||||
stage: KtxScanEnrichmentStage;
|
||||
input_hash: string;
|
||||
status: 'completed' | 'failed';
|
||||
output_json: string | null;
|
||||
error_message: string | null;
|
||||
updated_at: string;
|
||||
}
|
||||
|
||||
function parseStageRow<TOutput = unknown>(row: StageRow): KtxScanEnrichmentStageRecord<TOutput> {
|
||||
if (row.status === 'completed') {
|
||||
return {
|
||||
runId: row.run_id,
|
||||
connectionId: row.connection_id,
|
||||
syncId: row.sync_id,
|
||||
mode: row.mode,
|
||||
stage: row.stage,
|
||||
inputHash: row.input_hash,
|
||||
status: 'completed',
|
||||
output: JSON.parse(row.output_json ?? 'null') as TOutput,
|
||||
errorMessage: null,
|
||||
updatedAt: row.updated_at,
|
||||
};
|
||||
}
|
||||
function namespace(stage: KtxScanEnrichmentStage): string {
|
||||
return `scan:${stage}`;
|
||||
}
|
||||
|
||||
function metadataFor(input: {
|
||||
connectionId: string;
|
||||
syncId: string;
|
||||
mode: KtxScanMode;
|
||||
stage: KtxScanEnrichmentStage;
|
||||
}): Record<string, unknown> {
|
||||
return {
|
||||
runId: row.run_id,
|
||||
connectionId: row.connection_id,
|
||||
syncId: row.sync_id,
|
||||
mode: row.mode,
|
||||
stage: row.stage,
|
||||
inputHash: row.input_hash,
|
||||
status: 'failed',
|
||||
output: null,
|
||||
errorMessage: row.error_message ?? 'Unknown enrichment stage failure',
|
||||
updatedAt: row.updated_at,
|
||||
connectionId: input.connectionId,
|
||||
syncId: input.syncId,
|
||||
mode: input.mode,
|
||||
stage: input.stage,
|
||||
};
|
||||
}
|
||||
|
||||
function isSafeRunId(runId: string): boolean {
|
||||
return /^[a-zA-Z0-9][a-zA-Z0-9_.-]*$/.test(runId);
|
||||
function isScanMode(value: unknown): value is KtxScanMode {
|
||||
return typeof value === 'string' && (KTX_SCAN_MODES as readonly string[]).includes(value);
|
||||
}
|
||||
|
||||
const STAGES_TABLE = 'local_scan_enrichment_stages';
|
||||
const STAGES_PRIMARY_KEY = ['connection_id', 'stage', 'input_hash'] as const;
|
||||
function isScanEnrichmentStage(value: unknown): value is KtxScanEnrichmentStage {
|
||||
return typeof value === 'string' && (KTX_SCAN_ENRICHMENT_STAGES as readonly string[]).includes(value);
|
||||
}
|
||||
|
||||
function parseMetadata(record: ContentResultCacheRecord): ScanStageMetadata {
|
||||
const { connectionId, syncId, mode, stage } = record.metadata as Partial<ScanStageMetadata>;
|
||||
if (typeof connectionId !== 'string' || typeof syncId !== 'string' || !isScanMode(mode) || !isScanEnrichmentStage(stage)) {
|
||||
throw new Error(`Invalid scan enrichment cache metadata for ${record.namespace}/${record.scopeKey}`);
|
||||
}
|
||||
return { connectionId, syncId, mode, stage };
|
||||
}
|
||||
|
||||
function toScanRecord<TOutput = unknown>(record: ContentResultCacheRecord<TOutput>): KtxScanEnrichmentStageRecord<TOutput> {
|
||||
const metadata = parseMetadata(record);
|
||||
const base = {
|
||||
runId: record.runId,
|
||||
connectionId: metadata.connectionId,
|
||||
syncId: metadata.syncId,
|
||||
mode: metadata.mode,
|
||||
stage: metadata.stage,
|
||||
inputHash: record.inputHash,
|
||||
updatedAt: record.updatedAt,
|
||||
};
|
||||
if (record.status === 'completed') {
|
||||
return {
|
||||
...base,
|
||||
status: 'completed',
|
||||
output: record.output,
|
||||
errorMessage: null,
|
||||
};
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
status: 'failed',
|
||||
output: null,
|
||||
errorMessage: record.errorMessage,
|
||||
};
|
||||
}
|
||||
|
||||
export class SqliteLocalScanEnrichmentStateStore implements KtxScanEnrichmentStateStore {
|
||||
private readonly db: Database.Database;
|
||||
private readonly cache: ContentResultCache;
|
||||
|
||||
constructor(options: SqliteLocalScanEnrichmentStateStoreOptions) {
|
||||
mkdirSync(dirname(options.dbPath), { recursive: true });
|
||||
this.db = new Database(options.dbPath);
|
||||
this.db.pragma('journal_mode = WAL');
|
||||
// Disposable local resume cache: if a prior ktx wrote the table with a
|
||||
// different primary key, drop it rather than migrate. Losing it only means
|
||||
// one ingest cannot resume; it never corrupts a queryable artifact.
|
||||
this.dropStagesTableIfPrimaryKeyDiffers();
|
||||
this.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS local_scan_enrichment_stages (
|
||||
run_id TEXT NOT NULL,
|
||||
stage TEXT NOT NULL,
|
||||
input_hash TEXT NOT NULL,
|
||||
connection_id TEXT NOT NULL,
|
||||
sync_id TEXT NOT NULL,
|
||||
mode TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
output_json TEXT,
|
||||
error_message TEXT,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (connection_id, stage, input_hash)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS local_scan_enrichment_stages_content_idx
|
||||
ON local_scan_enrichment_stages (connection_id, stage, input_hash, updated_at);
|
||||
CREATE INDEX IF NOT EXISTS local_scan_enrichment_stages_run_idx
|
||||
ON local_scan_enrichment_stages (run_id, updated_at, stage);
|
||||
`);
|
||||
}
|
||||
|
||||
private dropStagesTableIfPrimaryKeyDiffers(): void {
|
||||
const columns = this.db.prepare(`PRAGMA table_info(${STAGES_TABLE})`).all() as Array<{
|
||||
name: string;
|
||||
pk: number;
|
||||
}>;
|
||||
if (columns.length === 0) {
|
||||
return;
|
||||
}
|
||||
const primaryKey = columns
|
||||
.filter((column) => column.pk > 0)
|
||||
.sort((left, right) => left.pk - right.pk)
|
||||
.map((column) => column.name);
|
||||
const matches =
|
||||
primaryKey.length === STAGES_PRIMARY_KEY.length &&
|
||||
primaryKey.every((name, index) => name === STAGES_PRIMARY_KEY[index]);
|
||||
if (!matches) {
|
||||
this.db.exec(`DROP TABLE ${STAGES_TABLE}`);
|
||||
}
|
||||
this.cache = options.cache ?? new SqliteContentResultCache({ dbPath: options.dbPath });
|
||||
}
|
||||
|
||||
async findCompletedStage<TOutput = unknown>(
|
||||
input: KtxScanEnrichmentStageLookup,
|
||||
): Promise<KtxScanEnrichmentCompletedStage<TOutput> | null> {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_scan_enrichment_stages
|
||||
WHERE connection_id = ?
|
||||
AND stage = ?
|
||||
AND input_hash = ?
|
||||
AND status = 'completed'
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 1
|
||||
`,
|
||||
)
|
||||
.get(input.connectionId, input.stage, input.inputHash) as StageRow | undefined;
|
||||
|
||||
if (!row) {
|
||||
return null;
|
||||
}
|
||||
const parsed = parseStageRow<TOutput>(row);
|
||||
return parsed.status === 'completed' ? parsed : null;
|
||||
const record = await this.cache.findCompletedResult<TOutput>({
|
||||
namespace: namespace(input.stage),
|
||||
scopeKey: input.connectionId,
|
||||
inputHash: input.inputHash,
|
||||
});
|
||||
return record ? (toScanRecord(record) as KtxScanEnrichmentCompletedStage<TOutput>) : null;
|
||||
}
|
||||
|
||||
async findLatestCompletedStage(input: {
|
||||
connectionId: string;
|
||||
stage: KtxScanEnrichmentStage;
|
||||
}): Promise<KtxScanEnrichmentCompletedStage | null> {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_scan_enrichment_stages
|
||||
WHERE connection_id = ?
|
||||
AND stage = ?
|
||||
AND status = 'completed'
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 1
|
||||
`,
|
||||
)
|
||||
.get(input.connectionId, input.stage) as StageRow | undefined;
|
||||
|
||||
if (!row) {
|
||||
return null;
|
||||
}
|
||||
const parsed = parseStageRow(row);
|
||||
return parsed.status === 'completed' ? parsed : null;
|
||||
const record = await this.cache.findLatestCompletedResult({
|
||||
namespace: namespace(input.stage),
|
||||
scopeKey: input.connectionId,
|
||||
});
|
||||
return record ? (toScanRecord(record) as KtxScanEnrichmentCompletedStage) : null;
|
||||
}
|
||||
|
||||
async saveCompletedStage<TOutput = unknown>(
|
||||
input: Omit<KtxScanEnrichmentCompletedStage<TOutput>, 'status' | 'errorMessage'>,
|
||||
): Promise<void> {
|
||||
this.db
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO local_scan_enrichment_stages (
|
||||
run_id,
|
||||
stage,
|
||||
input_hash,
|
||||
connection_id,
|
||||
sync_id,
|
||||
mode,
|
||||
status,
|
||||
output_json,
|
||||
error_message,
|
||||
updated_at
|
||||
)
|
||||
VALUES (
|
||||
@runId,
|
||||
@stage,
|
||||
@inputHash,
|
||||
@connectionId,
|
||||
@syncId,
|
||||
@mode,
|
||||
'completed',
|
||||
@outputJson,
|
||||
NULL,
|
||||
@updatedAt
|
||||
)
|
||||
ON CONFLICT(connection_id, stage, input_hash) DO UPDATE SET
|
||||
run_id = excluded.run_id,
|
||||
sync_id = excluded.sync_id,
|
||||
mode = excluded.mode,
|
||||
status = excluded.status,
|
||||
output_json = excluded.output_json,
|
||||
error_message = excluded.error_message,
|
||||
updated_at = excluded.updated_at
|
||||
`,
|
||||
)
|
||||
.run({
|
||||
runId: input.runId,
|
||||
stage: input.stage,
|
||||
inputHash: input.inputHash,
|
||||
connectionId: input.connectionId,
|
||||
syncId: input.syncId,
|
||||
mode: input.mode,
|
||||
outputJson: JSON.stringify(input.output),
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
await this.cache.saveCompletedResult({
|
||||
runId: input.runId,
|
||||
namespace: namespace(input.stage),
|
||||
scopeKey: input.connectionId,
|
||||
inputHash: input.inputHash,
|
||||
output: input.output,
|
||||
metadata: metadataFor(input),
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
}
|
||||
|
||||
async saveFailedStage(input: Omit<KtxScanEnrichmentFailedStage, 'status' | 'output'>): Promise<void> {
|
||||
this.db
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO local_scan_enrichment_stages (
|
||||
run_id,
|
||||
stage,
|
||||
input_hash,
|
||||
connection_id,
|
||||
sync_id,
|
||||
mode,
|
||||
status,
|
||||
output_json,
|
||||
error_message,
|
||||
updated_at
|
||||
)
|
||||
VALUES (
|
||||
@runId,
|
||||
@stage,
|
||||
@inputHash,
|
||||
@connectionId,
|
||||
@syncId,
|
||||
@mode,
|
||||
'failed',
|
||||
NULL,
|
||||
@errorMessage,
|
||||
@updatedAt
|
||||
)
|
||||
ON CONFLICT(connection_id, stage, input_hash) DO UPDATE SET
|
||||
run_id = excluded.run_id,
|
||||
sync_id = excluded.sync_id,
|
||||
mode = excluded.mode,
|
||||
status = excluded.status,
|
||||
output_json = excluded.output_json,
|
||||
error_message = excluded.error_message,
|
||||
updated_at = excluded.updated_at
|
||||
`,
|
||||
)
|
||||
.run({
|
||||
runId: input.runId,
|
||||
stage: input.stage,
|
||||
inputHash: input.inputHash,
|
||||
connectionId: input.connectionId,
|
||||
syncId: input.syncId,
|
||||
mode: input.mode,
|
||||
errorMessage: input.errorMessage,
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
await this.cache.saveFailedResult({
|
||||
runId: input.runId,
|
||||
namespace: namespace(input.stage),
|
||||
scopeKey: input.connectionId,
|
||||
inputHash: input.inputHash,
|
||||
errorMessage: input.errorMessage,
|
||||
metadata: metadataFor(input),
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
}
|
||||
|
||||
async listRunStages(runId: string): Promise<KtxScanEnrichmentStageRecord[]> {
|
||||
if (!isSafeRunId(runId)) {
|
||||
return [];
|
||||
}
|
||||
const rows = this.db
|
||||
.prepare(
|
||||
`
|
||||
SELECT *
|
||||
FROM local_scan_enrichment_stages
|
||||
WHERE run_id = ?
|
||||
ORDER BY updated_at ASC, stage ASC
|
||||
`,
|
||||
)
|
||||
.all(runId) as StageRow[];
|
||||
return rows.map((row) => parseStageRow(row));
|
||||
const records = await this.cache.listRunResults(runId);
|
||||
return records
|
||||
.filter((record) => record.namespace.startsWith('scan:'))
|
||||
.map((record) => toScanRecord(record));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,7 +10,9 @@ export type KtxConnectionDriver =
|
|||
| 'clickhouse'
|
||||
| 'mongodb';
|
||||
|
||||
export type KtxScanMode = 'structural' | 'relationships' | 'enriched';
|
||||
/** Canonical scan-mode registry. Runtime validation derives its allowlist here. */
|
||||
export const KTX_SCAN_MODES = ['structural', 'relationships', 'enriched'] as const;
|
||||
export type KtxScanMode = (typeof KTX_SCAN_MODES)[number];
|
||||
|
||||
export type KtxScanTrigger = 'cli' | 'mcp' | 'schema_scan' | 'scheduled' | 'manual';
|
||||
|
||||
|
|
|
|||
|
|
@ -135,6 +135,27 @@ export function slDeclaredSourceName(content: string): string | null {
|
|||
return typeof name === 'string' && name.length > 0 ? name : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Every standalone/overlay source file for a connection (excludes the `_schema/`
|
||||
* manifest). The one listing entry points share so a file is visible to all.
|
||||
*/
|
||||
export async function listSlSourceFiles(
|
||||
fileStore: Pick<KtxFileStorePort, 'listFiles' | 'readFile'>,
|
||||
connectionId: string,
|
||||
): Promise<SlSourceFile[]> {
|
||||
const dir = `semantic-layer/${assertSafeConnectionId(connectionId)}`;
|
||||
const schemaDir = `${dir}/_schema`;
|
||||
const listed = await fileStore.listFiles(dir);
|
||||
const paths = listed.files.filter((file) => isSlYamlPath(file) && !file.startsWith(`${schemaDir}/`)).sort();
|
||||
|
||||
const files: SlSourceFile[] = [];
|
||||
for (const path of paths) {
|
||||
const raw = await fileStore.readFile(path);
|
||||
files.push({ path, content: raw.content });
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
/**
|
||||
* Find the standalone/overlay file that defines `sourceName` for a connection.
|
||||
* Returns null when no file declares the name (the source may still exist as a
|
||||
|
|
@ -147,18 +168,9 @@ export async function resolveSlSourceFile(
|
|||
connectionId: string,
|
||||
sourceName: string,
|
||||
): Promise<SlSourceFile | null> {
|
||||
const dir = `semantic-layer/${assertSafeConnectionId(connectionId)}`;
|
||||
const schemaDir = `${dir}/_schema`;
|
||||
const listed = await fileStore.listFiles(dir);
|
||||
const paths = listed.files.filter((file) => isSlYamlPath(file) && !file.startsWith(`${schemaDir}/`)).sort();
|
||||
|
||||
const matches: SlSourceFile[] = [];
|
||||
for (const path of paths) {
|
||||
const raw = await fileStore.readFile(path);
|
||||
if (slSourceNameForFile(path, raw.content) === sourceName) {
|
||||
matches.push({ path, content: raw.content });
|
||||
}
|
||||
}
|
||||
const matches = (await listSlSourceFiles(fileStore, connectionId)).filter(
|
||||
(file) => slSourceNameForFile(file.path, file.content) === sourceName,
|
||||
);
|
||||
if (matches.length > 1) {
|
||||
throw new Error(
|
||||
`Multiple semantic-layer files declare source "${sourceName}": ${matches.map((match) => match.path).join(', ')}`,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue