Keep partial writes atomic, fill missing journal entries, and pick the true latest revision

apps/web/src/api/rpcClient.test.ts

@@ -0,0 +1,40 @@
+import { afterEach, expect, it, vi } from "vitest";
+import { rpc, rpcAsync } from "./rpcClient";
+
+afterEach(() => vi.unstubAllGlobals());
+
+const reason = "REFERENCE_INQUIRY_NOT_FOUND";
+const notice = `연구의 중간 기록은 저장됐습니다. 기준(reference) 단계만 실패했습니다 (사유: ${reason}). `
+  + "결과는 기록에서 볼 수 있습니다. 같은 질문을 새로 보내면 연구가 다시 실행될 수 있으니 먼저 저장된 결과를 확인하세요. / "
+  + `Intermediate research records were saved. Only the reference step failed (reason: ${reason}). `
+  + "You can read the result in the history. Sending the same question again as a new request may run the research again, so check the saved result first.";
+
+it("preserves the bilingual reference notice and saved-record metadata from an RPC error", async () => {
+  const data = { reason_code: reason, research_saved: true };
+  vi.stubGlobal("fetch", vi.fn().mockResolvedValue(Response.json({
+    jsonrpc: "2.0", id: "notice", error: { code: -32030, message: notice, data },
+  })));
+  await expect(rpc("thread/input", { project_id: "p" }, "reference-notice")).rejects.toMatchObject({
+    name: "RpcError", code: -32030, message: `${notice} [-32030]`, details: data,
+  });
+});
+
+it("shows the saved asynchronous operation's reference notice without replacing it", async () => {
+  const fetch = vi.fn()
+    .mockResolvedValueOnce(Response.json({
+      jsonrpc: "2.0", id: "started",
+      result: { operation_id: "op:notice", state: "RUNNING", value: { status: "ACCEPTED_RUNNING" } },
+    }))
+    .mockResolvedValueOnce(Response.json({
+      operation_id: "op:notice", state: "FAILED", result: null,
+      error: { code: -32603, message: notice, data: { reason_code: reason, research_saved: true } },
+    }));
+  vi.stubGlobal("fetch", fetch);
+  await expect(rpcAsync("thread/input", { project_id: "p", contract_version: 2 }, "async-notice", {
+    pollMs: 0,
+  })).rejects.toMatchObject({ message: notice });
+  expect(fetch).toHaveBeenCalledTimes(2);
+  expect(fetch).toHaveBeenLastCalledWith("/operations/op%3Anotice?project_id=p", {
+    credentials: "same-origin",
+  });
+});

config/module-responsibility-budget.json

@@ -123,7 +123,7 @@
       "policy": "over the 600-line limit when it was introduced; do not grow it; lower the baseline when it is split"
     },
     "src/thoth/application/commands/memory_full.py": {
-      "baseline_lines": 519,
+      "baseline_lines": 513,
       "top_level_symbols": [
         "MemoryFullHandlers"
       ],

docs/KNOWN_LIMITATIONS.ko.md

@@ -38,11 +38,24 @@ handler를 다시 실행하거나 과거 요청을 자동으로 훑어 닫지는
 있고, 새로 읽은 개정 번호로 보내면 같은 변경이 한 번 더 적용될 수도 있습니다. 이 복구가
 외부 효과의 정확히 한 번 실행을 보장하는 것은 아닙니다.
 
-## 끝난 요청의 후속 활동 기록이 빠질 수 있습니다
+## 같은 요청 재전송으로 빠진 성공 일지를 보충할 수 있습니다
 
-요청이 끝나면 결과를 먼저 저장하고, 활동 기록의 checkpoint·알림·"성공" 항목은 그 뒤에 저장합니다.
-둘 사이에 저장소 오류나 종료가 겹치면 결과와 연결 정보는 그대로이고 같은 요청은 그 결과를
-돌려주지만, 후속 항목은 빠질 수 있고 나중에 채워지지 않습니다. 요청을 다시 할 필요는 없습니다.
+로컬 모드에서는 실측한 20개 명령과 지원되는 저장 결과 형식에 한해, 같은 멱등 key를 다시 보내면
+빠진 checkpoint·알림·성공 기록을 채울 수 있습니다. 저장된 결과와 연결 정보는 그대로이며
+handler·모델·그 밖의 효과를 다시 실행하지 않습니다. 기록이 다 채워지면 재전송해도 추가하지 않습니다.
+
+새 항목의 시각은 실제 보충 시각이며, `supplemented: true`와 원 완료 시각을 담은
+`supplemented_for`가 붙습니다. 기존 항목은 고치지 않습니다. 보충 중 중단되면 그 거래의 새 기록은
+함께 취소되고 다음 같은 요청에서 다시 시도할 수 있습니다.
+
+심사용 모드, 실패·진행 중 요청, 연구 게시와 복원 게시에는 적용하지 않습니다. 다른 명령과
+확인하지 못한 옛 저장 형식도 보류합니다. 일지 형식이 불명확하거나 서로 충돌하는 경우, 시계 역행
+등으로 이벤트 체인의 정렬을 유지할 수 없는 경우, 작업 공간 소유권이 없거나 일시적인 보충 저장
+오류가 있는 경우에는 누락이 남을 수 있습니다. 기존 접근 권한 거절과 결과·binding 검증은 유지합니다.
+
+미측정 명령과 추가 source 분기 검토는 후속 작업입니다. 심사용·여러 프로세스 간 보충은 이번 변경
+범위 밖입니다. 일지만 고치려고 새 요청으로 다시 보내면 효과가 다시 실행될 수 있으니 저장된 결과를
+먼저 확인하세요.
 
 ## 샌드박스 연구 시험은 일부만 기록될 수 있습니다
 

docs/KNOWN_LIMITATIONS.md

@@ -40,13 +40,26 @@ its ordinary change notification is omitted. Check the saved result before sendi
 A new question may use model capacity again; a newly refreshed revision guard can also permit a
 second change. This recovery is not a guarantee of exactly-once external effects.
 
-## A finished request can miss some follow-up log entries
-
-When a request finishes, its result is saved first. The checkpoint, notification and "succeeded"
-entries of the activity log are saved afterwards. If a storage error or shutdown hits between the
-two, the result and its stored links stay intact and the same request returns that result, but
-those follow-up entries can be missing and are not filled in later. You do not need to repeat the
-request.
+## Missing success-log entries can be supplemented on retry
+
+In LOCAL mode, retrying the same idempotency key can fill missing checkpoint, notification and
+success entries for 20 measured commands and their supported saved result forms. The stored
+result and its links remain unchanged. The handler, model and other effects are not run again;
+once the missing entries are present, further retries add nothing.
+
+New entries use the actual supplementation time and carry `supplemented: true` and
+`supplemented_for` with the original completion time. Existing entries are not rewritten.
+An interrupted supplementation transaction rolls back its new entries and can be retried.
+
+This does not supplement hosted review, failed or running requests, research publication, or
+restore publication. Other methods and unmeasured legacy result forms remain deferred. Unknown
+or conflicting log formats, unsafe event-chain ordering (including clock rollback), missing
+workspace ownership or temporary repair-storage failure can leave the gap in place. Existing
+access denial and canonical result/binding checks still apply.
+
+Follow-up work includes the unmeasured methods and additional source-branch audits; hosted and
+cross-process supplementation are outside this change. Do not send a new request just to repair
+a log: it could run the effect again. Check the saved result first.
 
 ## Sandboxed research tests can leave a partial record
 

src/thoth/adapters/storage/action.py

@@ -92,7 +92,7 @@ def list_actions(self, project_id: str) -> tuple[ActionRecord, ...]:
             rows = connection.execute(
                 select(action_records)
                 .where(action_records.c.project_id == project_id)
-                .order_by(action_records.c.created_at, literal_column("rowid", Integer()))
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             values = tuple(ActionRecord.model_validate(load(row["content_json"])) for row in rows)
         latest: dict[str, ActionRecord] = {}
@@ -122,9 +122,7 @@ def read_action(
             action_records.c.action_id == action_id,
         )
         if revision_digest is None:
-            statement = statement.order_by(
-                action_records.c.created_at.desc(), literal_column("rowid", Integer()).desc()
-            ).limit(1)
+            statement = statement.order_by(literal_column("rowid", Integer()).desc()).limit(1)
         else:
             statement = statement.where(action_records.c.revision_digest == revision_digest)
         with read_connection(self._engine) as connection:
@@ -158,7 +156,7 @@ def list_portfolios(self, project_id: str) -> tuple[ActionPortfolioRecord, ...]:
             rows = connection.execute(
                 select(action_portfolios)
                 .where(action_portfolios.c.project_id == project_id)
-                .order_by(action_portfolios.c.created_at, literal_column("rowid", Integer()))
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             values = tuple(
                 ActionPortfolioRecord.model_validate(load(row["content_json"])) for row in rows
@@ -190,9 +188,7 @@ def read_portfolio(
             action_portfolios.c.portfolio_id == portfolio_id,
         )
         if revision_digest is None:
-            statement = statement.order_by(
-                action_portfolios.c.created_at.desc(), literal_column("rowid", Integer()).desc()
-            ).limit(1)
+            statement = statement.order_by(literal_column("rowid", Integer()).desc()).limit(1)
         else:
             statement = statement.where(action_portfolios.c.revision_digest == revision_digest)
         with read_connection(self._engine) as connection:
@@ -228,7 +224,7 @@ def list_plans(self, project_id: str) -> tuple[ActionPlanRecord, ...]:
             rows = connection.execute(
                 select(action_plans)
                 .where(action_plans.c.project_id == project_id)
-                .order_by(action_plans.c.created_at, literal_column("rowid", Integer()))
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             values = tuple(
                 ActionPlanRecord.model_validate(load(row["content_json"])) for row in rows
@@ -260,9 +256,7 @@ def read_plan(
             action_plans.c.plan_id == plan_id,
         )
         if revision_digest is None:
-            statement = statement.order_by(
-                action_plans.c.created_at.desc(), literal_column("rowid", Integer()).desc()
-            ).limit(1)
+            statement = statement.order_by(literal_column("rowid", Integer()).desc()).limit(1)
         else:
             statement = statement.where(action_plans.c.revision_digest == revision_digest)
         with read_connection(self._engine) as connection:

src/thoth/adapters/storage/control_record.py

@@ -4,7 +4,7 @@
 
 import orjson
 from pydantic import ValidationError
-from sqlalchemy import Engine, insert, select
+from sqlalchemy import Engine, Integer, insert, literal_column, select
 
 from thoth.adapters.storage.schema import control_records
 from thoth.adapters.storage.transaction import read_connection, write_connection
@@ -55,7 +55,9 @@ def read(self, project_id: str, namespace: str, record_id: str) -> ControlRecord
                         control_records.c.namespace == namespace,
                         control_records.c.record_id == record_id,
                     )
-                    .order_by(control_records.c.version.desc())
+                    .order_by(
+                        control_records.c.version.desc(), literal_column("rowid", Integer()).desc()
+                    )
                     .limit(1)
                 )
                 .mappings()
@@ -91,10 +93,13 @@ def list(
         )
         if record_type is not None:
             statement = statement.where(control_records.c.record_type == record_type)
+        order = (
+            (control_records.c.version, literal_column("rowid", Integer()))
+            if latest_only
+            else (control_records.c.created_at, control_records.c.version)
+        )
         with read_connection(self._engine) as connection:
-            rows = connection.execute(
-                statement.order_by(control_records.c.created_at, control_records.c.version)
-            ).mappings()
+            rows = connection.execute(statement.order_by(*order)).mappings()
             parsed: list[ControlRecord] = []
             for row in rows:
                 try:

src/thoth/adapters/storage/criterion_contract.py

@@ -140,7 +140,7 @@ def list_contracts(self, project_id: str) -> tuple[CriterionContractRecord, ...]
             rows = connection.execute(
                 select(criterion_contracts)
                 .where(criterion_contracts.c.project_id == project_id)
-                .order_by(criterion_contracts.c.created_at, literal_column("rowid", Integer()))
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             return self._latest(tuple(self._contract(row) for row in rows))
 
@@ -168,9 +168,7 @@ def read_contract(
         if revision_digest is not None:
             statement = statement.where(criterion_contracts.c.revision_digest == revision_digest)
         else:
-            statement = statement.order_by(
-                criterion_contracts.c.created_at.desc(), literal_column("rowid", Integer()).desc()
-            ).limit(1)
+            statement = statement.order_by(literal_column("rowid", Integer()).desc()).limit(1)
         with read_connection(self._engine) as connection:
             row = connection.execute(statement).mappings().first()
         return None if row is None else self._contract(row)

src/thoth/adapters/storage/decision_object.py

@@ -4,7 +4,7 @@
 from typing import cast
 
 import orjson
-from sqlalchemy import Engine, insert, select
+from sqlalchemy import Engine, Integer, insert, literal_column, select
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 
 from thoth.adapters.storage.schema import (
@@ -186,7 +186,7 @@ def list_objects(self, project_id: str) -> tuple[DecisionObjectRecord, ...]:
             rows = connection.execute(
                 select(decision_objects)
                 .where(decision_objects.c.project_id == project_id)
-                .order_by(decision_objects.c.created_at)
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             values = tuple(
                 DecisionObjectRecord.model_validate(load(row["content_json"])) for row in rows
@@ -209,7 +209,7 @@ def read_object(
             decision_objects.c.object_id == object_id,
         )
         if revision_digest is None:
-            statement = statement.order_by(decision_objects.c.created_at.desc()).limit(1)
+            statement = statement.order_by(literal_column("rowid", Integer()).desc()).limit(1)
         else:
             statement = statement.where(decision_objects.c.revision_digest == revision_digest)
         with read_connection(self._engine) as connection:
@@ -252,7 +252,7 @@ def list_relations(self, project_id: str, object_id: str) -> tuple[ObjectRelatio
                     object_relations.c.project_id == project_id,
                     object_relations.c.source_object_id == object_id,
                 )
-                .order_by(object_relations.c.created_at)
+                .order_by(literal_column("rowid", Integer()))
             ).mappings()
             values = tuple(
                 ObjectRelationRecord.model_validate(load(row["content_json"])) for row in rows
@@ -272,7 +272,7 @@ def read_relation(self, relation_id: str) -> ObjectRelationRecord | None:
                 connection.execute(
                     select(object_relations)
                     .where(object_relations.c.relation_id == relation_id)
-                    .order_by(object_relations.c.created_at.desc())
+                    .order_by(literal_column("rowid", Integer()).desc())
                     .limit(1)
                 )
                 .mappings()
@@ -315,7 +315,9 @@ def list_attention(
         if object_id is not None:
             statement = statement.where(object_attention.c.object_id == object_id)
         with read_connection(self._engine) as connection:
-            rows = connection.execute(statement.order_by(object_attention.c.created_at)).mappings()
+            rows = connection.execute(
+                statement.order_by(literal_column("rowid", Integer()))
+            ).mappings()
             values = tuple(
                 ObjectAttentionRecord.model_validate(load(row["content_json"])) for row in rows
             )
@@ -334,7 +336,7 @@ def read_attention(self, attention_id: str) -> ObjectAttentionRecord | None:
                 connection.execute(
                     select(object_attention)
                     .where(object_attention.c.attention_id == attention_id)
-                    .order_by(object_attention.c.created_at.desc())
+                    .order_by(literal_column("rowid", Integer()).desc())
                     .limit(1)
                 )
                 .mappings()

src/thoth/adapters/storage/events.py

@@ -1,15 +1,20 @@
 from __future__ import annotations
 
+from collections.abc import Generator
+from contextlib import contextmanager
 from datetime import datetime
 from typing import cast
 
 import orjson
-from sqlalchemy import Connection, Engine, insert, select
+from sqlalchemy import Connection, Engine, Integer, insert, literal_column, select
+from sqlalchemy.exc import OperationalError
 
 from thoth.adapters.storage.schema import checkpoints, events
 from thoth.adapters.storage.transaction import read_connection, write_connection
 from thoth.domain.event import Checkpoint, EventRecord
 from thoth.ports.event_store import EventStorePort
+from thoth.ports.journal import JournalCompletionUnavailable
+from thoth.ports.runtime import AtomicUnitOfWorkPort
 
 
 def _object(value: str) -> dict[str, object]:
@@ -40,6 +45,36 @@ class SqliteEventStore(EventStorePort):
     def __init__(self, engine: Engine) -> None:
         self._engine = engine
 
+    @contextmanager
+    def completion_transaction(self, uow: AtomicUnitOfWorkPort) -> Generator[None]:
+        try:
+            with uow.transaction():
+                yield
+        except OperationalError as exc:
+            raise JournalCompletionUnavailable("JOURNAL_STORAGE_UNAVAILABLE") from exc
+
+    def list_checkpoints(self, operation_id: str) -> tuple[Checkpoint, ...]:
+        with read_connection(self._engine) as connection:
+            rows = (
+                connection.execute(
+                    select(checkpoints)
+                    .where(checkpoints.c.operation_id == operation_id)
+                    .order_by(checkpoints.c.created_at, checkpoints.c.checkpoint_id)
+                )
+                .mappings()
+                .all()
+            )
+        return tuple(
+            Checkpoint(
+                checkpoint_id=str(row["checkpoint_id"]),
+                operation_id=str(row["operation_id"]),
+                payload=_object(str(row["payload_json"])),
+                checkpoint_digest=str(row["checkpoint_digest"]),
+                created_at=datetime.fromisoformat(str(row["created_at"])),
+            )
+            for row in rows
+        )
+
     def insert_event(self, event: EventRecord) -> None:
         with write_connection(self._engine) as connection:
             insert_event_on_connection(connection, event)
@@ -93,7 +128,7 @@ def latest_checkpoint(self, operation_id: str) -> Checkpoint | None:
                 connection.execute(
                     select(checkpoints)
                     .where(checkpoints.c.operation_id == operation_id)
-                    .order_by(checkpoints.c.created_at.desc(), checkpoints.c.checkpoint_id.desc())
+                    .order_by(literal_column("rowid", Integer()).desc())
                     .limit(1)
                 )
                 .mappings()

src/thoth/adapters/storage/evidence_graph.py

@@ -93,10 +93,7 @@ def read_source_by_artifact(self, artifact_id: str) -> EvidenceSourceRecord | No
                 connection.execute(
                     select(evidence_sources)
                     .where(evidence_sources.c.artifact_id == artifact_id)
-                    .order_by(
-                        evidence_sources.c.created_at.desc(),
-                        literal_column("rowid", Integer()).desc(),
-                    )
+                    .order_by(literal_column("rowid", Integer()).desc())
                     .limit(1)
                 )
                 .mappings()