Skip to content

Commit 3f5dbe8

Browse files
committed
Prevent historical knowledge backfill
1 parent 7a88d39 commit 3f5dbe8

15 files changed

Lines changed: 1548 additions & 11 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ pgdata/
105105
# Project-local generated artifacts
106106
.memory/
107107
artifacts/
108+
output/
108109
tests/visualization/artifacts/
109110

110111
# Local-only working directories

‎app/core/policies/episodes/knowledge_building.py‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,19 +16,26 @@ class KnowledgeBuildPlan:
1616

1717
episode_id: str
1818
trigger: KnowledgeBuildTrigger
19+
baseline_only: bool = False
1920

2021

2122
def plan_stable_watermark_builds(
2223
*,
2324
snapshots: Iterable[EpisodeBuildSnapshot],
2425
now: datetime,
2526
idle_stable_seconds: int,
27+
lifecycle_activated_at: datetime | None = None,
2628
) -> tuple[KnowledgeBuildPlan, ...]:
2729
"""Return build plans for closed or idle-stable episodes with new events."""
2830

2931
if idle_stable_seconds < 0:
3032
raise ValueError("idle_stable_seconds must be non-negative")
3133
stable_before = now - timedelta(seconds=idle_stable_seconds)
34+
legacy_baseline_before = (
35+
None
36+
if lifecycle_activated_at is None
37+
else lifecycle_activated_at - timedelta(seconds=idle_stable_seconds)
38+
)
3239
plans: list[KnowledgeBuildPlan] = []
3340
seen: set[str] = set()
3441
for snapshot in snapshots:
@@ -46,6 +53,10 @@ def plan_stable_watermark_builds(
4653
KnowledgeBuildPlan(
4754
episode_id=snapshot.episode_id,
4855
trigger=KnowledgeBuildTrigger.WATERMARK_STABLE,
56+
baseline_only=_should_baseline_legacy_first_build(
57+
snapshot=snapshot,
58+
legacy_baseline_before=legacy_baseline_before,
59+
),
4960
)
5061
)
5162
return tuple(plans)
@@ -69,3 +80,17 @@ def _is_stable_build_candidate(
6980
if snapshot.status is not EpisodeStatus.ACTIVE:
7081
return False
7182
return snapshot.latest_event_at <= stable_before
83+
84+
85+
def _should_baseline_legacy_first_build(
86+
*,
87+
snapshot: EpisodeBuildSnapshot,
88+
legacy_baseline_before: datetime | None,
89+
) -> bool:
90+
"""Return whether an old first-build episode should be watermarked only."""
91+
92+
if legacy_baseline_before is None:
93+
return False
94+
if snapshot.latest_successful_build_watermark is not None:
95+
return False
96+
return snapshot.latest_event_at <= legacy_baseline_before

‎app/core/ports/db/knowledge_builder.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,12 @@ def latest_successful_watermark(
2424
) -> int | None:
2525
"""Return the newest successfully processed event watermark."""
2626

27+
@abstractmethod
28+
def ensure_lifecycle_activation_started_at(
29+
self, *, repo_id: str, activated_at: datetime
30+
) -> datetime:
31+
"""Return the repo lifecycle-builder activation time, creating it if absent."""
32+
2733
@abstractmethod
2834
def list_running_runs(
2935
self, *, repo_id: str, episode_id: str

‎app/core/use_cases/knowledge_builder/build_knowledge/execute.py‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,16 @@ def execute_build_knowledge(
116116
finished_at=now,
117117
)
118118

119+
if request.baseline_only and previous_watermark is None:
120+
return _record_baseline_run(
121+
request=request,
122+
settings=settings,
123+
uow=uow,
124+
id_generator=id_generator,
125+
event_watermark=event_watermark,
126+
now=now,
127+
)
128+
119129
run_id = id_generator.new_id()
120130
uow.knowledge_build_runs.add(
121131
KnowledgeBuildRun(
@@ -261,6 +271,56 @@ def _fresh_running_run(
261271
return None
262272

263273

274+
def _record_baseline_run(
275+
*,
276+
request: BuildKnowledgeRequest,
277+
settings: BuildKnowledgeSettings,
278+
uow: IUnitOfWork,
279+
id_generator: IIdGenerator,
280+
event_watermark: int,
281+
now: datetime,
282+
) -> BuildKnowledgeResult:
283+
"""Persist a no-provider watermark for history that predates lifecycle builds."""
284+
285+
run_id = id_generator.new_id()
286+
run_summary = (
287+
"Baselined legacy episode without invoking build_knowledge; "
288+
"future events after this watermark remain eligible."
289+
)
290+
uow.knowledge_build_runs.add(
291+
KnowledgeBuildRun(
292+
id=run_id,
293+
repo_id=request.repo_id,
294+
episode_id=request.episode_id,
295+
trigger=request.trigger,
296+
status=KnowledgeBuildRunStatus.SKIPPED,
297+
event_watermark=event_watermark,
298+
previous_event_watermark=None,
299+
provider=settings.provider,
300+
model=settings.model,
301+
reasoning=settings.reasoning,
302+
run_summary=run_summary,
303+
error_code="legacy_episode_baselined",
304+
started_at=now,
305+
finished_at=now,
306+
created_at=now,
307+
)
308+
)
309+
return BuildKnowledgeResult(
310+
status=KnowledgeBuildRunStatus.SKIPPED,
311+
run_id=run_id,
312+
event_watermark=event_watermark,
313+
previous_event_watermark=None,
314+
provider=settings.provider,
315+
model=settings.model,
316+
reasoning=settings.reasoning,
317+
write_count=0,
318+
skipped_item_count=0,
319+
run_summary=run_summary,
320+
error_code="legacy_episode_baselined",
321+
)
322+
323+
264324
def _skipped_result(
265325
*,
266326
settings: BuildKnowledgeSettings,

‎app/core/use_cases/knowledge_builder/build_knowledge/request.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ class BuildKnowledgeRequest(_StrictModel):
1818
repo_root: str = Field(min_length=1)
1919
episode_id: str = Field(min_length=1)
2020
trigger: KnowledgeBuildTrigger
21+
baseline_only: bool = False
2122

2223
@field_validator("repo_id", "repo_root", "episode_id")
2324
@classmethod

‎app/infrastructure/db/runtime/models/knowledge_builder.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,20 @@
118118
),
119119
)
120120

121+
122+
knowledge_build_lifecycle = Table(
123+
"knowledge_build_lifecycle",
124+
metadata,
125+
Column("repo_id", String, primary_key=True),
126+
Column("activated_at", TIMESTAMP(timezone=True), nullable=False),
127+
Column(
128+
"created_at",
129+
TIMESTAMP(timezone=True),
130+
nullable=False,
131+
server_default=text("NOW()"),
132+
),
133+
)
134+
121135
Index(
122136
"idx_knowledge_build_runs_episode_status_created_at",
123137
knowledge_build_runs.c.repo_id,

‎app/infrastructure/db/runtime/repos/relational/knowledge_builder_repo.py‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,18 @@
55
from datetime import datetime, timezone
66

77
from sqlalchemy import desc, select, text, update
8+
from sqlalchemy.dialects.postgresql import insert
89

910
from app.core.entities.knowledge_builder import (
1011
KnowledgeBuildRun,
1112
KnowledgeBuildRunStatus,
1213
KnowledgeBuildTrigger,
1314
)
1415
from app.core.ports.db.knowledge_builder import IKnowledgeBuildRunsRepo
15-
from app.infrastructure.db.runtime.models.knowledge_builder import knowledge_build_runs
16+
from app.infrastructure.db.runtime.models.knowledge_builder import (
17+
knowledge_build_lifecycle,
18+
knowledge_build_runs,
19+
)
1620

1721

1822
class KnowledgeBuildRunsRepo(IKnowledgeBuildRunsRepo):
@@ -67,6 +71,27 @@ def latest_successful_watermark(
6771
)
6872
return None if row is None else int(row)
6973

74+
def ensure_lifecycle_activation_started_at(
75+
self, *, repo_id: str, activated_at: datetime
76+
) -> datetime:
77+
"""Return the repo lifecycle-builder activation time, creating it if absent."""
78+
79+
self._session.execute(
80+
insert(knowledge_build_lifecycle)
81+
.values(
82+
repo_id=repo_id,
83+
activated_at=activated_at,
84+
created_at=activated_at,
85+
)
86+
.on_conflict_do_nothing(index_elements=[knowledge_build_lifecycle.c.repo_id])
87+
)
88+
value = self._session.execute(
89+
select(knowledge_build_lifecycle.c.activated_at).where(
90+
knowledge_build_lifecycle.c.repo_id == repo_id
91+
)
92+
).scalar_one()
93+
return value
94+
7095
def list_running_runs(
7196
self, *, repo_id: str, episode_id: str
7297
) -> tuple[KnowledgeBuildRun, ...]:

‎app/infrastructure/process/episode_sync/poller.py‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -272,14 +272,22 @@ def _run_stable_builds_best_effort(
272272
return
273273
try:
274274
with uow_factory() as uow:
275+
now = _utc_now()
276+
lifecycle_activated_at = (
277+
uow.knowledge_build_runs.ensure_lifecycle_activation_started_at(
278+
repo_id=repo_id,
279+
activated_at=now,
280+
)
281+
)
275282
snapshots = tuple(uow.episodes.list_build_snapshots(repo_id=repo_id))
276283
except Exception:
277284
return
278285

279286
for plan in plan_stable_watermark_builds(
280287
snapshots=snapshots,
281-
now=_utc_now(),
288+
now=now,
282289
idle_stable_seconds=idle_stable_seconds,
290+
lifecycle_activated_at=lifecycle_activated_at,
283291
):
284292
_run_build_knowledge_plan_best_effort(
285293
run_build_knowledge=run_build_knowledge,
@@ -301,12 +309,15 @@ def _run_build_knowledge_plan_best_effort(
301309
if run_build_knowledge is None or plan is None:
302310
return
303311
try:
304-
run_build_knowledge(
305-
repo_id=repo_id,
306-
repo_root=repo_root,
307-
episode_id=plan.episode_id,
308-
trigger=plan.trigger,
309-
)
312+
kwargs = {
313+
"repo_id": repo_id,
314+
"repo_root": repo_root,
315+
"episode_id": plan.episode_id,
316+
"trigger": plan.trigger,
317+
}
318+
if plan.baseline_only:
319+
kwargs["baseline_only"] = True
320+
run_build_knowledge(**kwargs)
310321
except Exception:
311322
return
312323

‎app/startup/knowledge_builder.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ def run_build_knowledge(
2424
repo_root: Path,
2525
episode_id: str,
2626
trigger: KnowledgeBuildTrigger | str,
27+
baseline_only: bool = False,
2728
) -> dict[str, object]:
2829
"""Wire dependencies and run build_knowledge for one episode."""
2930

@@ -32,6 +33,7 @@ def run_build_knowledge(
3233
repo_root=str(repo_root.resolve()),
3334
episode_id=episode_id,
3435
trigger=KnowledgeBuildTrigger(trigger),
36+
baseline_only=baseline_only,
3537
)
3638
result = execute_build_knowledge(
3739
request,
Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
"""Add lifecycle activation state for forward-only knowledge building."""
2+
3+
from alembic import op
4+
5+
6+
revision = "20260520_0028"
7+
down_revision = "20260519_0027"
8+
branch_labels = None
9+
depends_on = None
10+
11+
12+
def upgrade() -> None:
13+
"""Create per-repo lifecycle activation state."""
14+
15+
op.execute(
16+
"""
17+
CREATE TABLE IF NOT EXISTS knowledge_build_lifecycle (
18+
repo_id TEXT PRIMARY KEY,
19+
activated_at TIMESTAMPTZ NOT NULL,
20+
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
21+
);
22+
"""
23+
)
24+
25+
26+
def downgrade() -> None:
27+
"""Drop per-repo lifecycle activation state."""
28+
29+
op.execute("DROP TABLE IF EXISTS knowledge_build_lifecycle;")

0 commit comments

Comments
 (0)