Close a stuck request on retry when its local owner process is gone

config/module-responsibility-budget.json

@@ -227,7 +227,7 @@
       "policy": "over the 600-line limit when it was introduced; do not grow it; lower the baseline when it is split"
     },
     "src/thoth/protocol/bus.py": {
-      "baseline_lines": 741,
+      "baseline_lines": 740,
       "top_level_symbols": [
         "DispatchTicket",
         "PreClaimAuthorization",

docs/KNOWN_LIMITATIONS.ko.md

@@ -20,15 +20,23 @@ English: [KNOWN_LIMITATIONS.md](KNOWN_LIMITATIONS.md)
 함께 쓰는 폴더나 클라우드 동기화·네트워크 공유로 여러 컴퓨터가 보는 작업 공간은 보호하지 못합니다.
 그런 방식으로 같은 작업 공간을 두 곳에서 열지 마세요. 심사용(hosted review) 모드는 이 잠금을 쓰지 않습니다.
 
-## 종료로 중단된 요청이 "실행 중"으로 남을 수 있습니다
-
-THOTH는 작업을 시작하기 전에 요청을 "실행 중"으로 기록합니다. 작업 도중 앱이 멈추면 그 기록이
-"실행 중"으로 남을 수 있습니다. 같은 요청을 같은 멱등 key로 다시 보내면 작업을 두 번 하지 않으려고
-다시 시작하지 않고 "실행 중"을 돌려줍니다. 이어서 하려면 새 질문으로 다시 하세요. 새 질문은
-모델 사용량을 다시 쓸 수 있으니, 먼저 이전 결과가 이미 저장됐는지 확인하세요.
-
-심사용(hosted review) 모드에서는 재시작 뒤 이런 요청이 "이전 호출의 종료를 확인하지 못했다"는
-안내와 함께 닫힐 수도 있습니다. 그 모드의 알려진 한계는 [VERIFICATION.md](VERIFICATION.md)에 있습니다.
+## 종료로 중단된 요청의 복구
+
+로컬 모드에서는 실측한 67개 명령에 한해, 같은 멱등 key를 다시 보내면 소유 worker의 종료가
+확인된 요청을 실패로 닫을 수 있습니다. 같은 호스트·부팅 식별과 이미 보유한 작업 공간 잠금도
+필요합니다. `OPERATION_OWNER_GONE`과 함께 저장된 내용을 확인하라는 안내를 반환하며,
+handler를 다시 실행하거나 과거 요청을 자동으로 훑어 닫지는 않습니다.
+
+목록 밖의 명령(연구 시작·입력·조정 및 일시정지·재개·중지 포함), 소유 정보가 없는 옛 기록,
+호스트 식별이 불명하거나 다른 경우, worker가 살아 있거나 생존 여부가 불명인 경우는
+실행 중으로 남을 수 있습니다. 시간 초과만으로 종료를 판단하지 않습니다. 심사용 모드의 복구
+정책은 그대로입니다. `improvement/evaluation/run`은 쓰기 없는 응답에 명시적인 unchanged 표시가
+없어, 별도 검토 전까지 허용 목록 밖의 후속 후보로 남습니다.
+
+봉인이 원래 완료와의 경합에서 이기면 도메인 변경은 이미 저장됐어도 일반 변경 알림은 빠질 수
+있습니다. 새 요청을 보내기 전에 저장된 결과를 확인하세요. 새 질문은 모델 사용량을 다시 쓸 수
+있고, 새로 읽은 개정 번호로 보내면 같은 변경이 한 번 더 적용될 수도 있습니다. 이 복구가
+외부 효과의 정확히 한 번 실행을 보장하는 것은 아닙니다.
 
 ## 끝난 요청의 후속 활동 기록이 빠질 수 있습니다
 

docs/KNOWN_LIMITATIONS.md

@@ -22,16 +22,23 @@ on a local disk. It does not cover a workspace folder shared between Windows and
 cloud sync client or a network share exposes to several computers; do not open the same workspace
 from two places that way. The hosted review mode does not use this lock.
 
-## A request interrupted by a shutdown can stay "running"
-
-THOTH records a request as running before it starts the work. If the application stops while the
-work is in progress, the record can stay "running". Sending the same request again with the same
-idempotency key returns "running" instead of starting the work a second time, so that nothing is
-done twice. To continue, ask again as a new question. A new question can use model capacity again,
-so check first whether the earlier result has already been saved.
-
-In the hosted review mode a restart can also close such a request with a message that the previous
-call could not be confirmed. See [VERIFICATION.md](VERIFICATION.md) for that mode's known limit.
+## Recovery of a request interrupted by shutdown
+
+In LOCAL mode, 67 measured commands can be closed as failed when the same idempotency key is
+sent again and the original worker is confirmed gone. THOTH also requires matching host/boot
+identity and an already-held workspace lock. It returns `OPERATION_OWNER_GONE` with guidance to
+check what was saved; it does not rerun the handler or scan old requests automatically.
+
+Requests outside that list (including research start/input/steer and pause/resume/stop), old
+records without ownership, an unavailable or different host identity, and a live or unknown
+worker can remain running. A timeout alone is not proof of death. The hosted review recovery
+policy is unchanged. `improvement/evaluation/run` remains outside the list pending separate
+review because its no-effect response lacks an explicit unchanged marker.
+
+If sealing wins a race with the original completion, a domain change may already be saved while
+its ordinary change notification is omitted. Check the saved result before sending a new request.
+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
 

migrations/env.py

@@ -8,6 +8,7 @@
 
 from thoth.adapters.storage import behavior_execution_schema as behavior_execution_schema
 from thoth.adapters.storage import evaluation_run_schema as evaluation_run_schema
+from thoth.adapters.storage import operation_owner_schema as operation_owner_schema
 from thoth.adapters.storage import research_schema as research_schema
 from thoth.adapters.storage import resource_scope_schema as resource_scope_schema
 from thoth.adapters.storage import test_validity_schema as test_validity_schema

migrations/versions/4c17f5089ea3_operation_owners.py

@@ -0,0 +1,24 @@
+"""Record internal operation process ownership without changing existing rows."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "4c17f5089ea3"
+down_revision = "3b06e4f87d92"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "operation_owners",
+        sa.Column("operation_id", sa.String(160), primary_key=True),
+        sa.Column("worker_id", sa.String(160), nullable=False),
+        sa.Column("process_id", sa.Integer(), nullable=False),
+        sa.Column("process_marker", sa.String(160), nullable=False),
+        sa.Column("host_token", sa.String(160), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("operation_owners")

src/thoth/adapters/operation_host_identity.py

@@ -0,0 +1,104 @@
+"""Best-effort host/boot equality token; not an authentication or liveness proof."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+import re
+import sys
+import threading
+from pathlib import Path
+from uuid import UUID
+
+from thoth.domain.deployment_mode import DeploymentMode
+
+_KEY = b"thoth.operation-owner.host.v1"
+_cache: tuple[int, str | None] | None = None
+_lock = threading.Lock()
+
+
+def _guid(value: str) -> str:
+    if not re.fullmatch(r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value):
+        raise ValueError("HOST_IDENTITY_FORMAT")
+    parsed = UUID(value)
+    if parsed.int == 0:
+        raise ValueError("HOST_IDENTITY_FORMAT")
+    return parsed.hex
+
+
+def _windows_identity() -> tuple[str, str]:
+    import winreg
+
+    access = winreg.KEY_READ | winreg.KEY_WOW64_64KEY
+    boot_path = (
+        r"SYSTEM\CurrentControlSet\Control\Session Manager\Memory Management\PrefetchParameters"
+    )
+
+    def boot() -> int:
+        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, boot_path, 0, access) as key:
+            value, kind = winreg.QueryValueEx(key, "BootId")
+        if kind != winreg.REG_DWORD or type(value) is not int or not 0 <= value <= 2**32 - 1:
+            raise ValueError("HOST_IDENTITY_FORMAT")
+        return value
+
+    before = boot()
+    with winreg.OpenKey(
+        winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Cryptography", 0, access
+    ) as key:
+        machine, kind = winreg.QueryValueEx(key, "MachineGuid")
+    after = boot()
+    if before != after or kind != winreg.REG_SZ or not isinstance(machine, str):
+        raise ValueError("HOST_IDENTITY_UNAVAILABLE")
+    return _guid(machine), "prefetch-bootid:" + str(before)
+
+
+def _bounded_text(path: str) -> str:
+    with Path(path).open("rb") as source:
+        raw = source.read(129)
+    if len(raw) > 128:
+        raise ValueError("HOST_IDENTITY_FORMAT")
+    return raw.decode("ascii").strip()
+
+
+def _linux_identity() -> tuple[str, str]:
+    machine = _bounded_text("/etc/machine-id")
+    if not re.fullmatch(r"[0-9a-fA-F]{32}", machine) or int(machine, 16) == 0:
+        raise ValueError("HOST_IDENTITY_FORMAT")
+    return machine.lower(), _guid(_bounded_text("/proc/sys/kernel/random/boot_id"))
+
+
+def _read_token() -> str | None:
+    try:
+        if sys.platform == "win32":
+            machine, boot = _windows_identity()
+        elif sys.platform == "linux":
+            machine, boot = _linux_identity()
+        else:
+            return None
+        payload = (sys.platform + "\0" + machine + "\0" + boot).encode("utf-8")
+        return hmac.new(_KEY, payload, hashlib.sha256).hexdigest()
+    except (OSError, ValueError, TypeError, ImportError):
+        return None
+
+
+def operation_host_token(mode: DeploymentMode) -> str | None:
+    """Read once per LOCAL process, caching failure too; HOSTED performs no lookup."""
+    if mode is not DeploymentMode.LOCAL:
+        return None
+    global _cache
+    with _lock:
+        pid = os.getpid()
+        if _cache is None or _cache[0] != pid:
+            _cache = (pid, _read_token())
+        return _cache[1]
+
+
+def _after_fork() -> None:
+    global _cache, _lock
+    _cache = None
+    _lock = threading.Lock()
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_after_fork)

src/thoth/adapters/storage/operation_owner_schema.py

@@ -0,0 +1,15 @@
+"""Internal process ownership, separate from the public operation columns."""
+
+from sqlalchemy import Column, Integer, String, Table
+
+from thoth.adapters.storage.schema import metadata
+
+operation_owners = Table(
+    "operation_owners",
+    metadata,
+    Column("operation_id", String(160), primary_key=True),
+    Column("worker_id", String(160), nullable=False),
+    Column("process_id", Integer, nullable=False),
+    Column("process_marker", String(160), nullable=False),
+    Column("host_token", String(160), nullable=False),
+)

src/thoth/adapters/storage/operation_owner_store.py

@@ -0,0 +1,30 @@
+"""Owner persistence on the caller's existing operation transaction."""
+
+from sqlalchemy import Connection, insert, select
+
+from thoth.adapters.storage.operation_owner_schema import operation_owners
+from thoth.domain.operation import OperationOwnerRecord
+
+
+def insert_owner(
+    connection: Connection, operation_id: str, owner: OperationOwnerRecord | None
+) -> None:
+    if owner is not None and owner.host_token is not None:
+        connection.execute(
+            insert(operation_owners).values(operation_id=operation_id, **owner.model_dump())
+        )
+
+
+def read_owner(connection: Connection, operation_id: str) -> OperationOwnerRecord | None:
+    row = (
+        connection.execute(
+            select(operation_owners).where(operation_owners.c.operation_id == operation_id)
+        )
+        .mappings()
+        .first()
+    )
+    if row is None:
+        return None
+    return OperationOwnerRecord.model_validate(
+        {key: row[key] for key in ("worker_id", "process_id", "process_marker", "host_token")}
+    )

src/thoth/adapters/storage/operations.py

@@ -13,6 +13,7 @@
 from sqlalchemy.exc import OperationalError
 
 from thoth.adapters.storage.events import insert_event_on_connection
+from thoth.adapters.storage.operation_owner_store import insert_owner, read_owner
 from thoth.adapters.storage.resource_scope_schema import operation_resource_bindings
 from thoth.adapters.storage.schema import (
     idempotency_keys,
@@ -150,6 +151,8 @@ def _claim(
         *,
         started_event: Callable[[], EventRecord] | None = None,
     ) -> tuple[OperationRecord, bool]:
+        if candidate.owner is not None and candidate.owner.host_token is None:
+            candidate = candidate.model_copy(update={"owner": None})
         # Logical operation bookkeeping owns its transaction; it must not borrow a
         # domain transaction that could later commit a partially failed claim.
         with self._engine.begin() as connection:
@@ -203,6 +206,7 @@ def _claim(
                     operation_id=candidate.operation_id,
                 )
             )
+            insert_owner(connection, candidate.operation_id, candidate.owner)
             if started_event is not None:
                 insert_event_on_connection(connection, started_event())
             return candidate, False
@@ -362,7 +366,10 @@ def _read_with_connection(
         )
         if row is None:
             return None
-        return self._with_resource_binding(connection, self._deserialize(row))
+        record = self._deserialize(row).model_copy(
+            update={"owner": read_owner(connection, operation_id)}
+        )
+        return self._with_resource_binding(connection, record)
 
     @staticmethod
     def _with_resource_binding(

src/thoth/adapters/storage/workspace_owner.py

@@ -14,6 +14,8 @@
 import os
 import sys
 import threading
+from collections.abc import Generator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
@@ -183,6 +185,26 @@ def acquire_workspace_owner(workspace: Path, *, command: str | None = None) -> W
         return WorkspaceOwnership(key)
 
 
+def holds_workspace(workspace: Path) -> bool:
+    with _GUARD:
+        return _key(workspace) in _OWNED
+
+
+@contextmanager
+def pin_workspace_owner(workspace: Path) -> Generator[bool]:
+    """Retain only an already-owned lock; never acquire or create a lock file."""
+    key = _key(workspace)
+    with _GUARD:
+        entry = _OWNED.get(key)
+        if entry is not None:
+            entry.count += 1
+    try:
+        yield entry is not None
+    finally:
+        if entry is not None:
+            _release(key)
+
+
 def _release(key: str) -> None:
     with _GUARD:
         entry = _OWNED.get(key)

src/thoth/apps/resource_scope_composition.py

@@ -23,13 +23,15 @@
 from thoth.ports.event_store import EventStorePort
 from thoth.ports.field_measurement import FieldMeasurementObserverPort
 from thoth.ports.governance import GovernanceStorePort
+from thoth.ports.journal import JournalPort
 from thoth.ports.ledger import ManagedLedgerPort
 from thoth.ports.operation import OperationStorePort
 from thoth.ports.project import ProjectStorePort
 from thoth.ports.runtime import ClockPort, IdGeneratorPort
 from thoth.ports.sandbox import SandboxPort
 from thoth.ports.store_bundle import StoreBundlePort
 from thoth.protocol.bus import CommandBus
+from thoth.protocol.operation_owner import OperationOwner
 from thoth.protocol.registry import MethodRegistry
 
 _LOCAL_SOURCE_KINDS = frozenset({"LOCAL", "GIT"})
@@ -113,6 +115,8 @@ def scope_bus(
     seal_abandoned_running: bool = False,
     before_continue: Callable[[str], Awaitable[None]] | None = None,
     queued_admission: Callable[[OperationRecord], dict[str, JsonValue] | None] | None = None,
+    owner: OperationOwner | None = None,
+    journal: JournalPort | None = None,
 ) -> CommandBus:
     handlers = OperationCommandHandlers(
         operations,
@@ -132,12 +136,13 @@ def scope_bus(
         operations,
         clock,
         ids,
-        journal=OperationJournal(events, clock, ids),
+        journal=journal if journal is not None else OperationJournal(events, clock, ids),
         measurement=measurement,
         resource_access=scopes,
         seal_abandoned_running=seal_abandoned_running,
         before_continue=before_continue,
         queued_admission=queued_admission,
+        owner=owner,
     )