diff --git a/README.ko.md b/README.ko.md index c7233ee..8fde066 100644 --- a/README.ko.md +++ b/README.ko.md @@ -151,6 +151,8 @@ Node.js가 필요하고 `%LOCALAPPDATA%\THOTH\tools`에만 설치하며, 전역 새 Windows 설치의 연구 데이터는 소스 폴더와 별개인 `%LOCALAPPDATA%\THOTH`에 저장됩니다. 이전 버전을 사용했다면 `serve --workspace "C:\기존경로\.thoth-local" --port 8765`로 원래 데이터 폴더를 지정하세요. 시작할 때 실제 저장 위치를 표시하며, 기존 자료를 자동 이동하지 않습니다. +작업 공간 하나는 한 번에 THOTH 프로세스 하나만 사용합니다. 같은 작업 공간에서 두 번째로 시작하면 +첫 번째를 닫을 때까지 안내 문구와 함께(종료 코드 3) 멈춥니다. 로그인은 THOTH 전용 프로필을 사용하며, 다른 프로그램의 기존 로그인 정보는 가져오지 않습니다. Claude 로그인은 그 프로필에서 공식 Claude Code를 실행하고, xAI는 기기 코드 로그인을 사용합니다. diff --git a/README.md b/README.md index a15fc77..b66989d 100644 --- a/README.md +++ b/README.md @@ -161,6 +161,8 @@ New Windows installations keep research data in `%LOCALAPPDATA%\THOTH`, independ downloaded source folder. If you used an earlier version, open its existing data directory with `serve --workspace "C:\path\to\your\old\.thoth-local" --port 8765`. THOTH reports the selected directory at startup and does not move old data automatically. +One workspace is used by one THOTH process at a time; a second start on the same workspace stops +with a message (exit code 3) until the first one is closed. Sign-in uses separate THOTH profiles; existing login information from other programs is not imported. Claude sign-in runs the official Claude Code executable in that profile, and xAI uses diff --git a/SOURCE_MANIFEST.json b/SOURCE_MANIFEST.json index 43ebbf8..065eddc 100644 --- a/SOURCE_MANIFEST.json +++ b/SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schema_version": "thoth-source-preview-manifest-v1", - "created_at": "2026-10-17T07:42:48.547432+00:00", + "created_at": "2026-10-18T07:15:13.948726+00:00", "license": "MIT", "git_history_included": false, "full_suite_verified": false, @@ -53,13 +53,13 @@ }, { "path": "README.ko.md", - "sha256": "90bba7e4018a3849fa6adc9466501bcdd00d24a21d1e3758ec54dc19f4ccebc7", - "bytes": 15087 + "sha256": "a78940c44f6de6de7e88c31888e9a142d2e9cf36f333f02e39fb017eb9c0c389", + "bytes": 15308 }, { "path": "README.md", - "sha256": "d32ff97ecb4231dd50722e9f7c288435a04419ecc5188c95bf92fd56d03b149b", - "bytes": 13223 + "sha256": "5f1ca35c11a5d6402775b966cbcfc8ab192d2df8b17fae66d1adb5888e605dee", + "bytes": 13380 }, { "path": "THIRD_PARTY_NOTICES.md", @@ -1878,7 +1878,7 @@ }, { "path": "config/module-responsibility-budget.json", - "sha256": "959ad3d047de944f630b12222fefb8c4aeaf9b8b11a030df6ddefa3d74e4c727", + "sha256": "428bab7a87160af210a40f3d8c5a8762aa588981bdbbc07cf53c8509f83b7195", "bytes": 8580 }, { @@ -1943,18 +1943,18 @@ }, { "path": "docs/INSTALL.md", - "sha256": "79779cd2081432e913e5a5d1da20c326d3264296ef4c01dfb386e5567f4c5d73", - "bytes": 17803 + "sha256": "1d13de828a5db4a5c42fcd1dc32d56541634ab81b4f96f4f50c293a15300aab7", + "bytes": 18422 }, { "path": "docs/KNOWN_LIMITATIONS.ko.md", - "sha256": "9bf3c8415df4111ca99c7ead35237141a280c72e6e999a2ce7fc18e149fd6d23", - "bytes": 6963 + "sha256": "84e4fe631a04704d90a354346240b016a273eedb5de9b78fe819e2ddb58be726", + "bytes": 7334 }, { "path": "docs/KNOWN_LIMITATIONS.md", - "sha256": "fc288d30d7cd1553eb0e0343551c5bb61ffb677ed1851b71bf22ec25aae1ce2a", - "bytes": 6115 + "sha256": "d7ff5b62fc5d3f0551165924b13889709d84b2778d8bf1cd2bcd9885cdcf807c", + "bytes": 6446 }, { "path": "docs/KNOWN_TEST_FAILURES.json", @@ -3423,8 +3423,8 @@ }, { "path": "src/thoth/adapters/http/app.py", - "sha256": "ae3ecc1ca1c17f1fd37715deb3873a31d10640e95e94a55c8232199ddd51411e", - "bytes": 20134 + "sha256": "0e73d5cf7c516af7d2e68c1bda3f27da818c7a5bb928f08425bb1653b5d12ca0", + "bytes": 20695 }, { "path": "src/thoth/adapters/http/async_routes.py", @@ -3968,8 +3968,8 @@ }, { "path": "src/thoth/adapters/storage/bundle.py", - "sha256": "838b335af3659b0da0bd691edd4386de07dcf86e2bab130d8965a82d26491073", - "bytes": 7565 + "sha256": "d336ad1fd13776c9028d6a8b6da7a3db14cb27a7b6da873f1302528d9bceac89", + "bytes": 8741 }, { "path": "src/thoth/adapters/storage/control_record.py", @@ -4206,6 +4206,11 @@ "sha256": "1873bb91ae61570a71ebc7a3510de7063e0c44d96374578a6e532c5cca156183", "bytes": 3077 }, + { + "path": "src/thoth/adapters/storage/workspace_owner.py", + "sha256": "39e1eb1688ea5ed18d658d7f4cb3dde7e1950d130111e0e213eeef628d963e42", + "bytes": 6348 + }, { "path": "src/thoth/adapters/storage/workspace_setup.py", "sha256": "bc4041bf09dd9403d27e6628acd3dd11cf3cc7fa797834b70398acc8d08c945e", @@ -4258,8 +4263,8 @@ }, { "path": "src/thoth/application/commands/action_mutation_projection.py", - "sha256": "9b14233578bfb26bd1996e774ddcf3bb1fec5e151bf572fc4c39aead76291431", - "bytes": 10690 + "sha256": "c8685a85ab929bcb994760f5d9d1f4442788404eeb8d8cbf7d69ac20cfe42934", + "bytes": 11037 }, { "path": "src/thoth/application/commands/action_read_projection.py", @@ -4278,8 +4283,8 @@ }, { "path": "src/thoth/application/commands/actions_full.py", - "sha256": "f57bb5320d67451e9ef2d949ed91e827bf3bd87aade5b62f483c08c0cf1fb0c0", - "bytes": 24473 + "sha256": "6fb9f529f80728d14ba40233708c36bd4570d5faa3d8e65d47e192c3b37c78ee", + "bytes": 24482 }, { "path": "src/thoth/application/commands/behavior_execution.py", @@ -4288,8 +4293,8 @@ }, { "path": "src/thoth/application/commands/closure_full.py", - "sha256": "14d6d865ee88ec31b3ca31b17c0a54ca9d286d8f3dda454b65e18d5143f0523e", - "bytes": 19117 + "sha256": "d52c4135ffaa6577691f6db2f8cc56065bfa0e1e93117cc5596edcac6fe66d64", + "bytes": 20416 }, { "path": "src/thoth/application/commands/criteria_full.py", @@ -4398,8 +4403,8 @@ }, { "path": "src/thoth/application/commands/investigation.py", - "sha256": "9dffc989fc7acb46d3668ecdc601c25110d9bc0a6fbb5e8e27f9c1d81bc957d9", - "bytes": 9244 + "sha256": "532f5187d99b83a19beff5c469e2eb5d93d53ae9e6830567a690bdf68aac2508", + "bytes": 10114 }, { "path": "src/thoth/application/commands/judgment_review.py", @@ -4588,13 +4593,13 @@ }, { "path": "src/thoth/application/commands/revision_publication_records.py", - "sha256": "9043090a4b89c5bf521177c29d58710a6ad325c8e67f6ef0ab1c322819de91b1", - "bytes": 6807 + "sha256": "e730da31c147ed332651363b5ed348beca2507f38d8ef5746060b3ce47701d5f", + "bytes": 7284 }, { "path": "src/thoth/application/commands/revision_publication_views.py", - "sha256": "f3875124ec08fa8319220ea1906a0e59a98b82531ebf4b4eb8d33a451e05e334", - "bytes": 2869 + "sha256": "edd3b90f8ba42ad972eff549dbd483264ff29d28424f61d9e3451e37db63d6b9", + "bytes": 3185 }, { "path": "src/thoth/application/commands/revision_read_projection.py", @@ -4608,8 +4613,8 @@ }, { "path": "src/thoth/application/commands/revisions_full.py", - "sha256": "a43626bf5c3f6c751b6d8af8e1c46e782b21f7ba6e953f4ca49b28db6824abd4", - "bytes": 43839 + "sha256": "47789fa59011ca3844bdc3404724cc8004f550cff979e029bcc63380df1864c8", + "bytes": 43546 }, { "path": "src/thoth/application/commands/revisions_inputs.py", @@ -4633,8 +4638,8 @@ }, { "path": "src/thoth/application/commands/threads.py", - "sha256": "4f35f68895303a83eb7a90369a70a30f949d6c62d86a54f9fd023527710c666a", - "bytes": 22112 + "sha256": "4335b93071fcba7d318304e7e1dbd239c0a573f1cbe8343e57af0561fa06b6a1", + "bytes": 23304 }, { "path": "src/thoth/application/commands/trace_closure.py", @@ -4753,8 +4758,8 @@ }, { "path": "src/thoth/application/services/authorization_consumption.py", - "sha256": "6349f523fd5d02496797e8f6d59c17329f14cb4604e26abe74846cad08ae3d3d", - "bytes": 8913 + "sha256": "7bfa561baa7b4d86aaac46b4243c6795ab99246937c629cd40880f51efa2012d", + "bytes": 9312 }, { "path": "src/thoth/application/services/authorization_currentness.py", @@ -5528,8 +5533,8 @@ }, { "path": "src/thoth/application/services/research_controls.py", - "sha256": "3f13300b03c786e0fa6de26c640ecd86a8caf59d0f41dd28aac56fd7f34c4df8", - "bytes": 7118 + "sha256": "cc125d31b07319b85253ba5896548f6a81635daa96fb55a5d3f4128fb5746cfe", + "bytes": 8846 }, { "path": "src/thoth/application/services/research_conversation_activity.py", @@ -6231,6 +6236,11 @@ "sha256": "65deaac5b69876bcfbac7ffa664dd1db492de8f9736cf799f0bbe8b0e24d9150", "bytes": 5553 }, + { + "path": "src/thoth/apps/workspace_guard.py", + "sha256": "a1006e536515fecb1be9a141854d7d01f0c5c386b02eccc29408102aa1c4a8a9", + "bytes": 1154 + }, { "path": "src/thoth/apps/workspace_paths.py", "sha256": "e476b133bdc0efdb21024818c2d50dab6fc8d3b0b22d217d3e0f4cde8f049216", @@ -6238,8 +6248,8 @@ }, { "path": "src/thoth/cli.py", - "sha256": "b4ff2d5076e9e116db72d524cecc1cf98ae3f5d0d6464544649931f44e19d5ca", - "bytes": 21963 + "sha256": "6781d09ba2bb520b6ff95123c117dbedbc155c9b741cf87ade2858811c1e87db", + "bytes": 22529 }, { "path": "src/thoth/domain/__init__.py", @@ -6528,8 +6538,8 @@ }, { "path": "src/thoth/domain/investigation.py", - "sha256": "1afdd0b553a32640cf0691db049add48f6881d610f651dfdf47c3ccd937b104d", - "bytes": 2541 + "sha256": "3458badfccb58ce2f6897cc4105498459cdfbc2bbdcd3587b79ddd34008783af", + "bytes": 2670 }, { "path": "src/thoth/domain/judgment_review.py", @@ -7203,8 +7213,8 @@ }, { "path": "src/thoth/protocol/notifications.py", - "sha256": "06fd2d9dbbe85ce8c53bd943551160547e48978e6f9039b962d29766fc91b570", - "bytes": 1661 + "sha256": "9e73efd0d3cdc951016daed135f43d49b51bc03b52cfbc825b85ee9bd54edc6e", + "bytes": 1959 }, { "path": "src/thoth/protocol/operation_claim.py", @@ -8026,6 +8036,11 @@ "sha256": "e65fdf805e31ae0c45d1754069a91a1974282bf42f3edac2a1666dd75207f801", "bytes": 5318 }, + { + "path": "tests/integration/repeat_probe.py", + "sha256": "260dcbbccb43314b7ed1f384c4490f0a03efe5682733ce1226c65a1fa01cc7f7", + "bytes": 5563 + }, { "path": "tests/integration/research_measurement_helpers.py", "sha256": "bc1444f58ab13d5ebcf76c073386d3e28190b8d6a0c22612d3d30bad3ebd4b86", @@ -9626,6 +9641,21 @@ "sha256": "467afe1e48b629b2f9447aa9e22dc3a19cce1ee579ecafe211bd875f1e28e228", "bytes": 6120 }, + { + "path": "tests/integration/test_stale_running_current_behavior.py", + "sha256": "b64169bb51afebab03c03497fa23b252049dff38954988e504cfe7e1a47d38ab", + "bytes": 13545 + }, + { + "path": "tests/integration/test_state_transition_noop.py", + "sha256": "0d4d00b2ba820024bd55a5f4a3441a25ce7b593909be305a2b208ef141ec2c72", + "bytes": 7234 + }, + { + "path": "tests/integration/test_state_transition_repeat_current_behavior.py", + "sha256": "1620bfa1c34d9c17cc5ed1477893596c720b097df597b99afa70f2e4ac919ba6", + "bytes": 18126 + }, { "path": "tests/integration/test_storage_migration_ddl_contract.py", "sha256": "55e246ca9efa5312baa9c7b39dc22361a13efd55680486a75cf62dbdca171c21", @@ -9746,6 +9776,16 @@ "sha256": "e6a5ef1002509cdaebae4236fa0df2a48470e6cc1996e0123c8331e583ca5c94", "bytes": 15163 }, + { + "path": "tests/integration/test_workspace_runtime_exclusivity_baseline.py", + "sha256": "11a3ea83a1afa6d44997238b13956e36f36073bdd9c6af6939abf1f9213c6c5e", + "bytes": 7958 + }, + { + "path": "tests/integration/test_workspace_runtime_lock.py", + "sha256": "dfc9c375c2e2eff25781411d6e2d1b5ecc16249ab4961c12e47aef98837ee5f4", + "bytes": 13669 + }, { "path": "tests/integration/test_xai_workspace_route.py", "sha256": "1aa50ddba2fde60b5c782eb7fab14820bf24041fbed75b7e145ba63d316b8280", @@ -9996,6 +10036,11 @@ "sha256": "17c50efc5b5f58ff7960cff2f06d7c942c58319afe73163a4baf90ce65fd0270", "bytes": 3200 }, + { + "path": "tests/unit/application/test_research_controls_repeat.py", + "sha256": "eff0576c17865c0f9110eeb3b710ec28cf551c8fbe815290c4376f4c42525460", + "bytes": 6133 + }, { "path": "tests/unit/application/test_research_final_producer_selection.py", "sha256": "88801b7a5de2c9646fc2b4dcbd8c42d404461dced2ce96d3c6d973047dbf059e", diff --git a/config/module-responsibility-budget.json b/config/module-responsibility-budget.json index 27d4dfa..52a5a7c 100644 --- a/config/module-responsibility-budget.json +++ b/config/module-responsibility-budget.json @@ -137,7 +137,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/revisions_full.py": { - "baseline_lines": 926, + "baseline_lines": 917, "top_level_symbols": [ "_authenticated_revision_provenance", "RevisionFullHandlers" diff --git a/docs/INSTALL.md b/docs/INSTALL.md index a796250..4ee09ff 100644 --- a/docs/INSTALL.md +++ b/docs/INSTALL.md @@ -213,6 +213,14 @@ Startup reports the chosen root and nearby legacy directories. Legacy detection current directory; it cannot discover every older ZIP extraction. Nothing is automatically moved or merged. An empty project list can mean a different workspace was selected. +One workspace is open in one THOTH process at a time. `serve`, `desktop`, `run`, `rpc`, `events`, +`run-pack` and the interactive shell take the workspace when they start. A second one on the same +workspace stops with exit code 3 and a Korean and English message that names the other process; +close that one and run it again. Commands that do not open the database (`workspace`, `doctor` +and the sign-in commands) still work next to a running server. The lock belongs to the process, so +a crash frees it. It is meant for a local disk; a folder shared between Windows and WSL or synced by +a cloud client is not covered. + | Stored item | What restarting preserves | | --- | --- | | Projects, sources, evidence, saved results and model preferences | Stored under the workspace. Reopen the same root after restarting the backend. A lost model connection does not hide existing local research. | diff --git a/docs/KNOWN_LIMITATIONS.ko.md b/docs/KNOWN_LIMITATIONS.ko.md index 73233fa..6344457 100644 --- a/docs/KNOWN_LIMITATIONS.ko.md +++ b/docs/KNOWN_LIMITATIONS.ko.md @@ -7,12 +7,14 @@ THOTH는 실험 단계의 소스 미리보기입니다. 이 문서는 저장된 English: [KNOWN_LIMITATIONS.md](KNOWN_LIMITATIONS.md) -## 작업 공간마다 THOTH는 하나만 실행하세요 - -작업 공간 하나에 THOTH 프로세스 하나를 실행하는 구성만 지원합니다. 같은 작업 공간에 두 번째를 -띄워도 막히지 않고, 두 프로세스가 동시에 기록하면 같은 버전 번호의 기록이 둘 생길 수 있습니다. -그러면 나중에 읽을 때 오래된 쪽이 보일 수 있습니다. 다른 THOTH가 켜져 있는지 확실하지 않으면 -먼저 닫고 이어서 하세요. +## 작업 공간마다 THOTH는 하나만 실행됩니다 + +작업 공간 하나는 한 번에 THOTH 프로세스 하나만 사용합니다. 같은 작업 공간에서 두 번째를 +시작하면 종료 코드 3과 함께 다른 프로세스를 알려 주는 한국어·영어 안내가 나오고, 이미 켜진 +서버가 요청을 받으면 기록하지 않고 "작업 공간 사용 중" 오류(HTTP 503)를 돌려줍니다. 다른 THOTH를 +닫은 뒤 다시 실행하세요. 이 확인은 로컬 디스크의 운영체제 파일 잠금에 기대므로, Windows와 WSL이 +함께 쓰는 폴더나 클라우드 동기화·네트워크 공유로 여러 컴퓨터가 보는 작업 공간은 보호하지 못합니다. +그런 방식으로 같은 작업 공간을 두 곳에서 열지 마세요. 심사용(hosted review) 모드는 이 잠금을 쓰지 않습니다. ## 종료로 중단된 요청이 "실행 중"으로 남을 수 있습니다 diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index f344d1e..e4781a1 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -8,12 +8,15 @@ verification state is in [VERIFICATION.md](VERIFICATION.md). 한국어 안내: [KNOWN_LIMITATIONS.ko.md](KNOWN_LIMITATIONS.ko.md) -## Use one THOTH process per workspace - -A workspace is supported with one running THOTH process. Starting a second one on the same -workspace is not blocked, and two processes writing at the same time can leave two records with -the same version number. A later read may then show the older one. If you are unsure whether -another instance is running, close it before you continue. +## One THOTH process per workspace is enforced + +A workspace is open in one THOTH process at a time. Starting a second one on the same workspace +stops with exit code 3 and a Korean and English message that names the other process, and a server +that is already running answers a request with a "workspace in use" error (HTTP 503) instead of +writing. Close the other one and run it again. The check relies on the operating system's file lock +on a local disk. It does not cover a workspace folder shared between Windows and WSL, or one that a +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" diff --git a/src/thoth/adapters/http/app.py b/src/thoth/adapters/http/app.py index c576b93..dba5627 100644 --- a/src/thoth/adapters/http/app.py +++ b/src/thoth/adapters/http/app.py @@ -29,6 +29,7 @@ is_hosted_credential_method, ) from thoth.adapters.http.internal_routes import mount_internal_routes +from thoth.adapters.storage.workspace_owner import WorkspaceInUse from thoth.apps.hosted_review_dispatch import hosted_dispatch from thoth.apps.hosted_review_snapshot import ( mark_workspace_live, @@ -431,7 +432,19 @@ def _resolve_bus(application: FastAPI) -> CommandBus: if existing is not None: return existing workspace = Path(os.environ.get("THOTH_WORKSPACE", ".thoth")) - runtime = create_runtime(workspace) + try: + runtime = create_runtime(workspace) + except WorkspaceInUse as error: + # Only reached when the server was not started through the CLI (which owns the workspace + # first). The path and process id are shown to the caller, never stored. + raise HTTPException( + status_code=503, + detail={ + "code": "WORKSPACE_IN_USE", + "message_ko": error.message_ko, + "message_en": error.message_en, + }, + ) from error application.state.runtime = runtime application.state.bus = runtime.bus if hosted_mode(): diff --git a/src/thoth/adapters/storage/bundle.py b/src/thoth/adapters/storage/bundle.py index 212f67c..7d14260 100644 --- a/src/thoth/adapters/storage/bundle.py +++ b/src/thoth/adapters/storage/bundle.py @@ -42,6 +42,8 @@ from thoth.adapters.storage.thread_runtime import SqliteThreadRuntimeStore from thoth.adapters.storage.threads import SqliteThreadStore from thoth.adapters.storage.transaction import SqliteAtomicUnitOfWork +from thoth.adapters.storage.workspace_owner import WorkspaceOwnership, acquire_workspace_owner +from thoth.domain.deployment_mode import DeploymentMode, parse_deployment_mode from thoth.ports.resource_scope import ( ResourceAccessPort, ResourceRecordAccessPort, @@ -52,6 +54,7 @@ class SqliteStoreBundle: def __init__(self, ledger: SqliteLedger, workspace: Path) -> None: self.ledger = ledger + self._ownership: WorkspaceOwnership | None = None self.objects = ContentAddressedObjectStore(workspace) self.atomic_uow = SqliteAtomicUnitOfWork(ledger.engine) self.immediate_uow = SqliteAtomicUnitOfWork(ledger.engine, immediate=True) @@ -129,8 +132,17 @@ def evidence_uow( self.ledger.engine, resource_scopes=resource_scopes, fault_injector=fault_injector ) + def hold_workspace(self, ownership: WorkspaceOwnership | None) -> None: + """Release the workspace together with this bundle.""" + self._ownership = ownership + def close(self) -> None: - self.ledger.close() + try: + self.ledger.close() + finally: + ownership, self._ownership = self._ownership, None + if ownership is not None: + ownership.release() def scoped_ledger(self, resource_access: ResourceAccessPort) -> ScopedLedger: return ScopedLedger(self.ledger, resource_access) @@ -141,12 +153,26 @@ class SqliteStoreFactory: version = "1.0.0" def open(self, workspace: Path) -> SqliteStoreBundle: - database = workspace / "db" / "thoth.sqlite3" - migrate_sqlite_database(database) - ledger = SqliteLedger(database) + # One THOTH process per local workspace; the hosted review mode keeps one runtime per + # session workspace by its own deployment rule and is not locked here. + ownership = ( + None + if parse_deployment_mode() is DeploymentMode.HOSTED_REVIEW + else acquire_workspace_owner(workspace) + ) try: - ledger.initialize() - return SqliteStoreBundle(ledger, workspace) + database = workspace / "db" / "thoth.sqlite3" + migrate_sqlite_database(database) + ledger = SqliteLedger(database) + try: + ledger.initialize() + bundle = SqliteStoreBundle(ledger, workspace) + bundle.hold_workspace(ownership) + return bundle + except BaseException: + ledger.close() + raise except BaseException: - ledger.close() + if ownership is not None: + ownership.release() raise diff --git a/src/thoth/adapters/storage/workspace_owner.py b/src/thoth/adapters/storage/workspace_owner.py new file mode 100644 index 0000000..70e0eba --- /dev/null +++ b/src/thoth/adapters/storage/workspace_owner.py @@ -0,0 +1,205 @@ +"""One THOTH process per workspace. + +The process that opens a workspace's database takes an exclusive lock on byte 0 of +`/runtime.lock` (Windows `msvcrt.locking`, POSIX `fcntl.flock`). The operating system +releases it when the process ends, however it ends, so a leftover file never blocks a restart. +Runtimes inside the same process share the one lock by reference count. What the lock file says +(process id, start time, command) sits after byte 0 and is only a hint shown to the person who is +refused; it is never written to the database, the event journal or the diagnostics. +""" + +from __future__ import annotations + +import json +import os +import sys +import threading +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import BinaryIO + +LOCK_FILE = "runtime.lock" +_INFO_LIMIT = 4096 + + +@dataclass(frozen=True) +class WorkspaceHolder: + process_id: int | None + started_at: str | None + command: str | None + + +class WorkspaceInUse(RuntimeError): + """Another THOTH process already has this workspace open.""" + + def __init__(self, workspace: Path, holder: WorkspaceHolder | None) -> None: + self.workspace = workspace + self.holder = holder + super().__init__("WORKSPACE_IN_USE") + + def _detail(self, korean: bool) -> str: + holder = self.holder + if holder is None: + return "" + parts: list[str] = [] + if holder.process_id is not None: + parts.append(f"PID {holder.process_id}") + if holder.started_at is not None: + parts.append(f"{holder.started_at} 시작" if korean else f"started {holder.started_at}") + if holder.command: + parts.append(f"명령: {holder.command}" if korean else f"command: {holder.command}") + return f" ({', '.join(parts)})" if parts else "" + + @property + def message_ko(self) -> str: + return ( + f"이 작업 공간은 이미 다른 THOTH가 사용 중입니다{self._detail(True)}. " + f"그 THOTH를 먼저 종료한 뒤 다시 실행하세요. 작업 공간: {self.workspace}" + ) + + @property + def message_en(self) -> str: + return ( + f"This workspace is already in use by another THOTH{self._detail(False)}. " + f"Close that one first, then run this again. Workspace: {self.workspace}" + ) + + +class WorkspaceOwnership: + """One reference to the process-wide ownership of a workspace.""" + + def __init__(self, key: str) -> None: + self._key = key + self._released = False + + def release(self) -> None: + if self._released: + return + self._released = True + _release(self._key) + + +@dataclass +class _Entry: + count: int + stream: BinaryIO + + +_OWNED: dict[str, _Entry] = {} +_GUARD = threading.Lock() + + +def _key(workspace: Path) -> str: + return os.path.normcase(str(workspace.resolve())) + + +def _lock_byte(stream: BinaryIO) -> bool: + try: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + return False + + +def _unlock_byte(stream: BinaryIO) -> None: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + + +def _own_command() -> str: + name = Path(sys.argv[0]).name if sys.argv and sys.argv[0] else "thoth" + sub = sys.argv[1] if len(sys.argv) > 1 else "" + return f"{name} {sub}".strip() if sub.isidentifier() or sub.replace("-", "").isalpha() else name + + +def _write_info(stream: BinaryIO, command: str | None) -> None: + info = { + "pid": os.getpid(), + "started_at": datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC"), + "command": command or _own_command(), + } + stream.seek(1) + stream.truncate() + stream.write(json.dumps(info).encode("utf-8")) + stream.flush() + + +def _read_holder(path: Path) -> WorkspaceHolder | None: + try: + with path.open("rb") as stream: + stream.seek(1) + raw = stream.read(_INFO_LIMIT) + value = json.loads(raw.decode("utf-8")) + pid, started, command = value.get("pid"), value.get("started_at"), value.get("command") + return WorkspaceHolder( + process_id=pid if isinstance(pid, int) else None, + started_at=started if isinstance(started, str) else None, + command=command if isinstance(command, str) else None, + ) + except (OSError, ValueError, AttributeError): + return None + + +def acquire_workspace_owner(workspace: Path, *, command: str | None = None) -> WorkspaceOwnership: + """Own the workspace for this process, or raise WorkspaceInUse naming the other process.""" + root = workspace.resolve() + key = _key(root) + with _GUARD: + entry = _OWNED.get(key) + if entry is not None: + entry.count += 1 + return WorkspaceOwnership(key) + root.mkdir(parents=True, exist_ok=True) + path = root / LOCK_FILE + descriptor = os.open(path, os.O_RDWR | os.O_CREAT | getattr(os, "O_BINARY", 0), 0o600) + stream = os.fdopen(descriptor, "r+b") + try: + if stream.seek(0, os.SEEK_END) == 0: + stream.write(b"\0") + stream.flush() + if not _lock_byte(stream): + raise WorkspaceInUse(root, _read_holder(path)) + _write_info(stream, command) + except BaseException: + stream.close() + raise + _OWNED[key] = _Entry(count=1, stream=stream) + return WorkspaceOwnership(key) + + +def _release(key: str) -> None: + with _GUARD: + entry = _OWNED.get(key) + if entry is None: + return + entry.count -= 1 + if entry.count > 0: + return + del _OWNED[key] + try: + entry.stream.seek(1) + entry.stream.truncate() + entry.stream.flush() + except OSError: + pass + finally: + try: + _unlock_byte(entry.stream) + finally: + entry.stream.close() diff --git a/src/thoth/application/commands/action_mutation_projection.py b/src/thoth/application/commands/action_mutation_projection.py index 625ff0e..e978e8a 100644 --- a/src/thoth/application/commands/action_mutation_projection.py +++ b/src/thoth/application/commands/action_mutation_projection.py @@ -236,12 +236,18 @@ def authorization_prepare_result( } -def authorization_decide_result(decided: AuthorizationEnvelopeRecord) -> dict[str, JsonValue]: - return { +def authorization_decide_result( + decided: AuthorizationEnvelopeRecord, before: AuthorizationEnvelopeRecord | None = None +) -> dict[str, JsonValue]: + result: dict[str, JsonValue] = { "authorization": decided.model_dump(mode="json"), "exact_scope_digest": decided.exact_scope_digest, "single_use": decided.single_use, } + if before is not None and before.revision_digest == decided.revision_digest: + # The same role had already cast this vote: nothing was written. + result.update(unchanged=True, unchanged_reason="AUTHORIZATION_VOTE_ALREADY_RECORDED") + return result def compensation_specification(request: CompensationCreateInput) -> dict[str, object]: diff --git a/src/thoth/application/commands/actions_full.py b/src/thoth/application/commands/actions_full.py index 22289a7..713acff 100644 --- a/src/thoth/application/commands/actions_full.py +++ b/src/thoth/application/commands/actions_full.py @@ -416,7 +416,7 @@ async def authorization_decide(self, value: dict[str, JsonValue]) -> dict[str, J ) except ValueError as exc: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, str(exc)) from exc - return authorization_decide_result(decided) + return authorization_decide_result(decided, current) async def compensation_create(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: request = CompensationCreateInput.model_validate(value) diff --git a/src/thoth/application/commands/closure_full.py b/src/thoth/application/commands/closure_full.py index 49a7a77..b4ba1f3 100644 --- a/src/thoth/application/commands/closure_full.py +++ b/src/thoth/application/commands/closure_full.py @@ -18,6 +18,16 @@ from thoth.ports.runtime import AtomicUnitOfWorkPort from thoth.protocol.jsonrpc import RpcApplicationError, RpcErrorCode +# The decision or reopening was already recorded: answer with it and write nothing. +_UNCHANGED_DECIDED: dict[str, JsonValue] = { + "unchanged": True, + "unchanged_reason": "CLOSURE_ALREADY_DECIDED", +} +_UNCHANGED_REOPENED: dict[str, JsonValue] = { + "unchanged": True, + "unchanged_reason": "CLOSURE_ALREADY_REOPENED", +} + class ProjectInput(DomainModel): project_id: str = Field(min_length=1, max_length=160) @@ -285,6 +295,18 @@ async def decide(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: readiness = self._records.read(request.project_id, "CLOSURE", request.readiness_id) if readiness is None or readiness.record_type != "READINESS": raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, "readiness not found") + outcome = "DECIDED" if request.decision == "DECIDE" else "REJECTED" + existing = next( + ( + item + for item in self._records.list(request.project_id, "CLOSURE", "DECISION") + if item.payload.get("readiness_id") == readiness.record_id + and item.state == outcome + ), + None, + ) + if existing is not None: + return {"closure": existing.model_dump(mode="json"), **_UNCHANGED_DECIDED} if request.decision == "DECIDE" and readiness.state == "NOT_READY": raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, "closure is blocked") current_payload, _ = self._assess_readiness( @@ -310,7 +332,7 @@ async def decide(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: project_id=request.project_id, namespace="CLOSURE", record_type="DECISION", - state="DECIDED" if request.decision == "DECIDE" else "REJECTED", + state=outcome, payload={ "readiness_id": readiness.record_id, "closure_scope": readiness.payload.get("closure_scope"), @@ -358,6 +380,17 @@ def _reopen(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: parent = self._records.read(request.project_id, "CLOSURE", request.closure_id) if parent is None or parent.record_type != "DECISION" or parent.state != "DECIDED": raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, "CLOSURE_DECISION_REQUIRED") + existing = next( + ( + item + for item in self._records.list(request.project_id, "CLOSURE", "REOPEN") + if item.payload.get("closure_id") == request.closure_id + and item.payload.get("child_scope_ref") == request.child_scope_ref + ), + None, + ) + if existing is not None: + return {"reopen": existing.model_dump(mode="json"), **_UNCHANGED_REOPENED} record = self._controls.create( project_id=request.project_id, namespace="CLOSURE", diff --git a/src/thoth/application/commands/investigation.py b/src/thoth/application/commands/investigation.py index 89677bf..7c59fcc 100644 --- a/src/thoth/application/commands/investigation.py +++ b/src/thoth/application/commands/investigation.py @@ -2,7 +2,10 @@ from pydantic import Field, JsonValue -from thoth.application.services.investigation_service import InvestigationService +from thoth.application.services.investigation_service import ( + TERMINAL_STATE_BY_STOP_REASON, + InvestigationService, +) from thoth.domain.base import DomainModel from thoth.domain.investigation import InvestigationRecord from thoth.ports.acquisition import AcquisitionTraceStorePort @@ -11,6 +14,15 @@ from thoth.protocol.jsonrpc import RpcApplicationError, RpcErrorCode +def _unchanged(current: InvestigationRecord, reason: str) -> dict[str, JsonValue]: + """The state was already in place: answer with it and write nothing.""" + return { + "investigation": current.model_dump(mode="json"), + "unchanged": True, + "unchanged_reason": reason, + } + + class InvestigationListInput(DomainModel): project_id: str = Field(min_length=1, max_length=160) thread_id: str | None = Field(default=None, max_length=160) @@ -177,6 +189,13 @@ async def update(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: async def pause(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: request = InvestigationReadInput.model_validate(value) current = self._read(request) + if current.is_ended or current.execution_state == "PAUSED": + return _unchanged( + current, + "INVESTIGATION_ALREADY_ENDED" + if current.is_ended + else "INVESTIGATION_ALREADY_PAUSED", + ) try: updated = self._service.pause(current) except ValueError as exc: @@ -198,6 +217,9 @@ async def resume(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: async def stop(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: request = InvestigationStopInput.model_validate(value) current = self._read(request) + if current.is_ended and request.reason in TERMINAL_STATE_BY_STOP_REASON: + # An ended investigation keeps its verdict and result; a later stop never replaces them. + return _unchanged(current, "INVESTIGATION_ALREADY_ENDED") try: updated = self._service.stop(current, reason=request.reason) except ValueError as exc: diff --git a/src/thoth/application/commands/revision_publication_records.py b/src/thoth/application/commands/revision_publication_records.py index 66c1a78..8ebbeac 100644 --- a/src/thoth/application/commands/revision_publication_records.py +++ b/src/thoth/application/commands/revision_publication_records.py @@ -2,6 +2,8 @@ from __future__ import annotations +from collections.abc import Iterable + from thoth.application.commands.revisions_inputs import ( BaselineDecideInput, ChangeSetCommitInput, @@ -23,6 +25,21 @@ ) +def current_baseline_for( + baselines: Iterable[ControlRecord], candidate: ControlRecord +) -> ControlRecord | None: + """The CURRENT baseline that was already made from this candidate, if there is one.""" + return next( + ( + item + for item in baselines + if item.state == "CURRENT" + and item.payload.get("candidate_digest") == candidate.record_digest + ), + None, + ) + + def domain_transition_names(entity_type: str, content: dict[str, object]) -> list[str]: domain_transitions: list[str] = [] if entity_type == "ACTION" and content.get("proposal_state") == "RETIRED": diff --git a/src/thoth/application/commands/revision_publication_views.py b/src/thoth/application/commands/revision_publication_views.py index 153f455..952ab94 100644 --- a/src/thoth/application/commands/revision_publication_views.py +++ b/src/thoth/application/commands/revision_publication_views.py @@ -80,6 +80,15 @@ def typed_baseline_response( } +def unchanged_baseline_response(baseline: ControlRecord) -> dict[str, JsonValue]: + """The candidate already is the current baseline: answer with it and write nothing.""" + return { + **baseline_response(baseline), + "unchanged": True, + "unchanged_reason": "BASELINE_ALREADY_DECIDED", + } + + def baseline_response(baseline: ControlRecord) -> dict[str, JsonValue]: return cast( dict[str, JsonValue], diff --git a/src/thoth/application/commands/revisions_full.py b/src/thoth/application/commands/revisions_full.py index 56cfd18..88c402e 100644 --- a/src/thoth/application/commands/revisions_full.py +++ b/src/thoth/application/commands/revisions_full.py @@ -720,22 +720,7 @@ async def baseline_decide(self, value: dict[str, JsonValue]) -> dict[str, JsonVa None, ) if typed_candidate is not None: - role = next( - ( - item - for item in self._governance.list_roles(request.project_id) - if item.role_assignment_id == request.role_assignment_ref - and item.actor_id == request.actor_ref - and item.role == "baseline-owner" - and item.state == "ACTIVE" - ), - None, - ) - if role is None: - raise RpcApplicationError( - RpcErrorCode.DOMAIN_REJECTED, - "baseline authority role missing", - ) + self._require_baseline_owner(request) try: decision, baseline = self._baselines.decide( project_id=request.project_id, @@ -755,21 +740,7 @@ async def baseline_decide(self, value: dict[str, JsonValue]) -> dict[str, JsonVa raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, "candidate not found") if candidate.record_digest != request.approved_digest: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, "approved digest mismatch") - role = next( - ( - item - for item in self._governance.list_roles(request.project_id) - if item.role_assignment_id == request.role_assignment_ref - and item.actor_id == request.actor_ref - and item.role == "baseline-owner" - and item.state == "ACTIVE" - ), - None, - ) - if role is None: - raise RpcApplicationError( - RpcErrorCode.DOMAIN_REJECTED, "baseline authority role missing" - ) + self._require_baseline_owner(request) if request.decision == "REJECT": rejected = self._controls.create( project_id=request.project_id, @@ -782,6 +753,9 @@ async def baseline_decide(self, value: dict[str, JsonValue]) -> dict[str, JsonVa return {"baseline_candidate": rejected.model_dump(mode="json")} with self._ledger.transaction(): prior = self._records.list(request.project_id, "REVISION", "BASELINE") + same = publication.current_baseline_for(prior, candidate) + if same is not None: + return publication_views.unchanged_baseline_response(same) for item in prior: if item.state == "CURRENT": self._controls.create( @@ -801,6 +775,23 @@ async def baseline_decide(self, value: dict[str, JsonValue]) -> dict[str, JsonVa ) return publication_views.baseline_response(baseline) + def _require_baseline_owner(self, request: BaselineDecideInput) -> None: + role = next( + ( + item + for item in self._governance.list_roles(request.project_id) + if item.role_assignment_id == request.role_assignment_ref + and item.actor_id == request.actor_ref + and item.role == "baseline-owner" + and item.state == "ACTIVE" + ), + None, + ) + if role is None: + raise RpcApplicationError( + RpcErrorCode.DOMAIN_REJECTED, "baseline authority role missing" + ) + def _record(self, request: RecordReadInput, record_type: str) -> ControlRecord: item = self._records.read(request.project_id, "REVISION", request.record_id) if item is None or item.record_type != record_type: diff --git a/src/thoth/application/commands/threads.py b/src/thoth/application/commands/threads.py index a0b5751..072bb35 100644 --- a/src/thoth/application/commands/threads.py +++ b/src/thoth/application/commands/threads.py @@ -62,6 +62,21 @@ class ThreadMetadataUpdateInput(ThreadReadInput): display_name: str = Field(min_length=1, max_length=500) +# The state was already in place: the request answers with it and writes nothing. +_UNCHANGED_PAUSE: dict[str, JsonValue] = { + "unchanged": True, + "unchanged_reason": "THREAD_ALREADY_PAUSED", +} +_UNCHANGED_RESUME: dict[str, JsonValue] = { + "unchanged": True, + "unchanged_reason": "THREAD_ALREADY_RESUMED", +} +_UNCHANGED_STOP: dict[str, JsonValue] = { + "unchanged": True, + "unchanged_reason": "THREAD_ALREADY_STOPPED", +} + + class ThreadCommandHandlers: def __init__( self, @@ -284,6 +299,16 @@ async def pause(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: with self._ledger.transaction(): request = ThreadReadInput.model_validate(value) thread = self._read_scoped(request) + if thread.execution_state in { + ThreadExecutionState.PAUSED, + ThreadExecutionState.PAUSE_PENDING, + }: + checkpoints = self._runtime.list_checkpoints(request.project_id, request.thread_id) + return { + **self._serialize(thread), + "checkpoint": checkpoints[-1].model_dump(mode="json") if checkpoints else None, + **_UNCHANGED_PAUSE, + } target = ( ThreadExecutionState.PAUSE_PENDING if thread.execution_state == ThreadExecutionState.RUNNING @@ -323,6 +348,8 @@ async def resume(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: RpcErrorCode.STALE_CHECKPOINT, "thread resume rejected because checkpoint is missing or stale", ) + if thread.execution_state == ThreadExecutionState.IDLE: + return {**self._serialize(thread), **_UNCHANGED_RESUME} updated = thread.model_copy( update={ "execution_state": ThreadExecutionState.IDLE, @@ -338,6 +365,8 @@ async def stop(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: with self._ledger.transaction(): request = ThreadReadInput.model_validate(value) thread = self._read_scoped(request) + if thread.lifecycle == ThreadLifecycle.STOPPED: + return {**self._serialize(thread), **_UNCHANGED_STOP} updated = thread.model_copy( update={ "lifecycle": ThreadLifecycle.STOPPED, diff --git a/src/thoth/application/services/authorization_consumption.py b/src/thoth/application/services/authorization_consumption.py index 68e3b4f..865d0a4 100644 --- a/src/thoth/application/services/authorization_consumption.py +++ b/src/thoth/application/services/authorization_consumption.py @@ -178,6 +178,14 @@ def decide_authorization( raise ValueError("actor lacks a required non-fungible role assignment") if decision not in {"APPROVE", "REJECT"}: raise ValueError("authorization decision must be APPROVE or REJECT") + if any( + vote.get("role_assignment_ref") == role_assignment_ref + and vote.get("actor_ref") == actor_ref + and vote.get("decision") == decision + and vote.get("approved_digest") == approved_digest + for vote in current.decision_history + ): + return current # the same role already cast this vote: nothing to write now = clock.now() history: dict[str, object] = { "decision": decision, diff --git a/src/thoth/application/services/research_controls.py b/src/thoth/application/services/research_controls.py index 9424db3..54bbc6a 100644 --- a/src/thoth/application/services/research_controls.py +++ b/src/thoth/application/services/research_controls.py @@ -61,23 +61,29 @@ async def pause(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: lease = self.host.records.journal_read(project, f"lease:{thread_id}", ResearchLease) active = lease is not None and lease.state == "HELD" state = ThreadExecutionState.PAUSE_PENDING if active else ThreadExecutionState.PAUSED - if not self.host.threads.update( - thread.model_copy( - update={"execution_state": state, "revision": thread.revision + 1} - ), - expected_revision=thread.revision, - ): - raise ValueError("THREAD_REVISION_CHANGED") status = "PAUSE_REQUESTED" if active else "PAUSED" - self.host.records.journal( - project, operation.operation_id, attempt.model_copy(update={"status": status}) - ) + # Already in the requested pause (thread and attempt both): write nothing. + repeated = thread.execution_state == state and attempt.status == status + if not repeated: + if not self.host.threads.update( + thread.model_copy( + update={"execution_state": state, "revision": thread.revision + 1} + ), + expected_revision=thread.revision, + ): + raise ValueError("THREAD_REVISION_CHANGED") + self.host.records.journal( + project, operation.operation_id, attempt.model_copy(update={"status": status}) + ) return { "project_id": project, "thread_id": thread_id, "operation_id": operation.operation_id, "execution_state": state.value, "pause_accepted": True, + **( + {"unchanged": True, "unchanged_reason": "THREAD_ALREADY_PAUSED"} if repeated else {} + ), } async def resume(self, value: dict[str, JsonValue]) -> dict[str, JsonValue] | AcceptedRunning: @@ -115,6 +121,25 @@ async def resume(self, value: dict[str, JsonValue]) -> dict[str, JsonValue] | Ac lease = self.host.records.journal_read( operation.project_id, f"lease:{thread_id}", ResearchLease ) + if ( + attempt.status == "RUNNING" + and lease is not None + and lease.state == "HELD" + and lease.expires_at > self.host.records.clock.now() + and self.host.leases.worker.is_alive( + lease.process_id, lease.process_marker, lease.worker_id + ) + is True + ): + # Already running under a live worker: no write and no second dispatch. (A dead, + # unknown or expired holder still takes the recovery path below.) + return { + **self.host.accepted_value(attempt), + "resume_allowed": True, + "execution_state": thread.execution_state.value, + "unchanged": True, + "unchanged_reason": "THREAD_ALREADY_RESUMED", + } state = ( ThreadExecutionState.RUNNING if lease is not None and lease.state == "HELD" @@ -141,6 +166,13 @@ async def stop(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: result = await self.host.legacy.stop(value) if found is not None: attempt, operation = found + if result.get("unchanged") is True: + if operation.state != OperationState.RUNNING: + result["cancelled"] = False # this request cancelled nothing new + return result + # Stopped thread, request still running (a first stop was cut off before it + # cancelled): finish that cancel now, as a normal stop. + del result["unchanged"], result["unchanged_reason"] self.host.operations.cancel( operation.operation_id, completed_at=self.host.records.clock.now() ) diff --git a/src/thoth/apps/workspace_guard.py b/src/thoth/apps/workspace_guard.py new file mode 100644 index 0000000..aab6c83 --- /dev/null +++ b/src/thoth/apps/workspace_guard.py @@ -0,0 +1,38 @@ +"""Command-line guard for the one-process-per-workspace rule.""" + +from __future__ import annotations + +from collections.abc import Generator +from contextlib import contextmanager +from pathlib import Path + +import typer + +from thoth.adapters.storage.workspace_owner import WorkspaceInUse, acquire_workspace_owner + +WORKSPACE_IN_USE_EXIT_CODE = 3 + + +@contextmanager +def workspace_guard() -> Generator[None]: + """A workspace another THOTH has open is refused with one Korean and one English line.""" + try: + yield + except WorkspaceInUse as error: + typer.echo(error.message_ko, err=True) + typer.echo(error.message_en, err=True) + raise typer.Exit(code=WORKSPACE_IN_USE_EXIT_CODE) from error + + +@contextmanager +def owned_workspace(workspace: Path, *, command: str) -> Generator[None]: + """Own the workspace for a long-running command and let go when it ends. + + A server opens its runtime on the first request, which is too late to refuse a second one. + """ + with workspace_guard(): + ownership = acquire_workspace_owner(workspace, command=command) + try: + yield + finally: + ownership.release() diff --git a/src/thoth/cli.py b/src/thoth/cli.py index 63595ef..cc096f3 100644 --- a/src/thoth/cli.py +++ b/src/thoth/cli.py @@ -37,6 +37,7 @@ from thoth.apps.projectpack_execution import run_project_pack from thoth.apps.runtime import create_runtime from thoth.apps.tui_presentation import present_research_progress, render_tui_turn +from thoth.apps.workspace_guard import owned_workspace, workspace_guard from thoth.apps.workspace_paths import ( default_workspace, legacy_workspace_candidates, @@ -86,11 +87,12 @@ def main( if context.invoked_subcommand is not None: return announce_workspace(workspace, command="thoth") - _interactive_shell( - workspace, - sandbox_profile=sandbox_profile, - connector_config=connector_config, - ) + with workspace_guard(): + _interactive_shell( + workspace, + sandbox_profile=sandbox_profile, + connector_config=connector_config, + ) def _interactive_shell( @@ -339,13 +341,14 @@ def run_request( ] = None, ) -> None: """Execute one JSON-RPC request and emit one machine-readable JSON result.""" - runtime = create_runtime( - workspace, - sandbox_adapter=_sandbox_adapter(sandbox_profile, workspace), - connector_registry=( - None if connector_config is None else load_connector_registry(connector_config) - ), - ) + with workspace_guard(): + runtime = create_runtime( + workspace, + sandbox_adapter=_sandbox_adapter(sandbox_profile, workspace), + connector_registry=( + None if connector_config is None else load_connector_registry(connector_config) + ), + ) try: async def run_and_finish() -> bytes: @@ -395,13 +398,14 @@ def rpc_stdio( ] = None, ) -> None: """Serve JSON-RPC over stdin/stdout JSONL until EOF.""" - runtime = create_runtime( - workspace, - sandbox_adapter=_sandbox_adapter(sandbox_profile, workspace), - connector_registry=( - None if connector_config is None else load_connector_registry(connector_config) - ), - ) + with workspace_guard(): + runtime = create_runtime( + workspace, + sandbox_adapter=_sandbox_adapter(sandbox_profile, workspace), + connector_registry=( + None if connector_config is None else load_connector_registry(connector_config) + ), + ) try: async def serve() -> None: @@ -432,7 +436,8 @@ def events( "input": {"project_id": project_id, "operation_id": operation_id}, }, } - runtime = create_runtime(workspace) + with workspace_guard(): + runtime = create_runtime(workspace) try: response = asyncio.run(handle_json_line(orjson.dumps(request), runtime.bus)) finally: @@ -466,9 +471,10 @@ def run_pack( if selected_provider == "scripted" else create_codex_model(pack_workspace, model_name) ) - result = asyncio.run( - run_project_pack(loaded, workspace=pack_workspace, model=selected_model) - ) + with workspace_guard(): + result = asyncio.run( + run_project_pack(loaded, workspace=pack_workspace, model=selected_model) + ) except ModelExecutionHold as exc: typer.echo(str(exc), err=True) raise typer.Exit(code=1) from exc @@ -492,10 +498,11 @@ def serve( if host not in {"127.0.0.1", "localhost", "::1"}: raise typer.BadParameter("V0 local server only permits loopback hosts") announce_workspace(workspace, command="thoth serve") - os.environ["THOTH_WORKSPACE"] = str(workspace.resolve()) - from thoth.adapters.http import app as local_app + with owned_workspace(workspace, command="thoth serve"): + os.environ["THOTH_WORKSPACE"] = str(workspace.resolve()) + from thoth.adapters.http import app as local_app - uvicorn.run(local_app, host=host, port=port, reload=False) + uvicorn.run(local_app, host=host, port=port, reload=False) @app.command("hosted-review-serve") @@ -554,38 +561,39 @@ def desktop( if host not in {"127.0.0.1", "localhost", "::1"}: raise typer.BadParameter("V0 local server only permits loopback hosts") announce_workspace(workspace, command="thoth desktop") - os.environ["THOTH_WORKSPACE"] = str(workspace.resolve()) - if web_ui_dist() is None: - console.print("apps/web/dist 없음. `npm --prefix apps/web run build` 후 다시 실행.") - from thoth.adapters.http import app as local_app - - url = f"http://{host}:{port}/" - thread = threading.Thread( - target=lambda: uvicorn.run(local_app, host=host, port=port, reload=False), - daemon=True, - ) - thread.start() - health = f"http://{host}:{port}/healthz" - for _ in range(50): - try: - urllib.request.urlopen(health, timeout=0.4) - break - except (OSError, urllib.error.URLError): - time.sleep(0.1) - else: - console.print("desktop API did not become ready on loopback") - raise typer.Exit(code=1) - command = desktop_window_command( - url, - system=platform.system(), - program_files=Path(os.environ.get("PROGRAMFILES", r"C:\Program Files")), - program_files_x86=Path(os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)")), - ) - if command is not None: - subprocess.Popen(command) - else: - webbrowser.open(url) - thread.join() + with owned_workspace(workspace, command="thoth desktop"): + os.environ["THOTH_WORKSPACE"] = str(workspace.resolve()) + if web_ui_dist() is None: + console.print("apps/web/dist 없음. `npm --prefix apps/web run build` 후 다시 실행.") + from thoth.adapters.http import app as local_app + + url = f"http://{host}:{port}/" + thread = threading.Thread( + target=lambda: uvicorn.run(local_app, host=host, port=port, reload=False), + daemon=True, + ) + thread.start() + health = f"http://{host}:{port}/healthz" + for _ in range(50): + try: + urllib.request.urlopen(health, timeout=0.4) + break + except (OSError, urllib.error.URLError): + time.sleep(0.1) + else: + console.print("desktop API did not become ready on loopback") + raise typer.Exit(code=1) + command = desktop_window_command( + url, + system=platform.system(), + program_files=Path(os.environ.get("PROGRAMFILES", r"C:\Program Files")), + program_files_x86=Path(os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)")), + ) + if command is not None: + subprocess.Popen(command) + else: + webbrowser.open(url) + thread.join() if __name__ == "__main__": diff --git a/src/thoth/domain/investigation.py b/src/thoth/domain/investigation.py index dc28091..a621e01 100644 --- a/src/thoth/domain/investigation.py +++ b/src/thoth/domain/investigation.py @@ -55,6 +55,10 @@ def enforce_investigation_contract(self) -> InvestigationRecord: raise ValueError("terminal investigation requires result") return self + @property + def is_ended(self) -> bool: + return self.domain_state in {"CONVERGED", "HOLD", "ABSTAINED", "STOPPED"} + class InvestigationAuditRecord(DomainModel): audit_id: str diff --git a/src/thoth/protocol/notifications.py b/src/thoth/protocol/notifications.py index 9cf69f5..e36c8ae 100644 --- a/src/thoth/protocol/notifications.py +++ b/src/thoth/protocol/notifications.py @@ -22,6 +22,11 @@ def notifications_for(method: str, result: dict[str, JsonValue]) -> tuple[str, ...]: + # "unchanged": True means the request found the state already in place: nothing was written, + # so nothing is announced. (A thread/stop result's "cancelled" then means "this request newly + # cancelled something", so it is False.) + if result.get("unchanged") is True: + return () values = list(STATIC_NOTIFICATIONS.get(method, ())) append_research_notifications(method, result, values) append_decision_notifications(method, result, values) diff --git a/tests/integration/repeat_probe.py b/tests/integration/repeat_probe.py new file mode 100644 index 0000000..1587b11 --- /dev/null +++ b/tests/integration/repeat_probe.py @@ -0,0 +1,125 @@ +"""Measurement helper for the state-transition repeat tests (V-S): not a test. + +`RepeatProbe` wraps `CommandBus.dispatch`. After a target command succeeded inside an existing flow +test it sends the same input again under a NEW idempotency key and records what that repeat did: +whether it was refused, which domain tables it changed, and which notifications it emitted. The +database is then restored from a backup taken before the repeat, so the flow under test never sees +the probe (same method as the V-B measurement). A repeat under a new key is what a person meets +when they press a button twice or re-send after an interruption. +""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Awaitable, Callable +from contextlib import closing +from dataclasses import dataclass, field +from typing import Any + +import pytest +from tests.integration.storage_coverage_helpers import domain_snapshot + +from thoth.protocol.bus import CommandBus +from thoth.protocol.jsonrpc import JsonRpcRequest, JsonRpcResponse + + +@dataclass(frozen=True) +class Repeat: + method: str + accepted: bool + error_message: str | None + changed_tables: frozenset[str] + added_rows: dict[str, int] + notifications: tuple[str, ...] + unchanged_reason: str | None = None + cancelled: object = None + + +@dataclass +class RepeatProbe: + methods: frozenset[str] + # Optionally change the repeated input in place (e.g. another decision or scope). + change: Callable[[str, dict[str, Any]], None] | None = None + results: dict[str, list[Repeat]] = field(default_factory=dict) # type: ignore[type-arg] + _active: bool = False + + def install(self, monkeypatch: pytest.MonkeyPatch) -> None: + original = CommandBus.dispatch + probe = self + + async def dispatch(bus: CommandBus, request: JsonRpcRequest) -> JsonRpcResponse: + response = await original(bus, request) + if ( + probe._active # pyright: ignore[reportPrivateUsage] + or request.method not in probe.methods + or response.error is not None + ): + return response + probe._active = True # pyright: ignore[reportPrivateUsage] + try: + probe.results.setdefault(request.method, []).append( + await probe._repeat(bus, original, request, probe.change) # pyright: ignore[reportPrivateUsage] + ) + finally: + probe._active = False # pyright: ignore[reportPrivateUsage] + return response + + monkeypatch.setattr(CommandBus, "dispatch", dispatch) + + @staticmethod + async def _repeat( + bus: CommandBus, + original: Callable[[CommandBus, JsonRpcRequest], Awaitable[JsonRpcResponse]], + request: JsonRpcRequest, + change: Callable[[str, dict[str, Any]], None] | None, + ) -> Repeat: + engine = bus._operations._engine # pyright: ignore[reportPrivateUsage,reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownVariableType] + database = str(engine.url.database) # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType,reportUnknownArgumentType] + before = domain_snapshot(engine) # pyright: ignore[reportUnknownArgumentType] + saved = sqlite3.connect(":memory:") + with closing(sqlite3.connect(database)) as live: + live.backup(saved) + raw = request.model_dump(mode="json", by_alias=True) + key = request.params.meta.idempotency_key + "#repeat" + raw["id"] = key + raw["params"]["_meta"]["idempotencyKey"] = key + if change is not None: + change(request.method, raw["params"]["input"]) + outcome = await original(bus, JsonRpcRequest.model_validate(raw)) + after = domain_snapshot(engine) # pyright: ignore[reportUnknownArgumentType] + notifications: tuple[str, ...] = () + unchanged_reason: str | None = None + cancelled: object = None + if outcome.result is not None: + answered = outcome.result.get("value") + if isinstance(answered, dict): + if answered.get("unchanged") is True: + unchanged_reason = str(answered.get("unchanged_reason")) + cancelled = answered.get("cancelled") + with closing(sqlite3.connect(database)) as live: + rows = live.execute( + "SELECT event_type FROM events WHERE operation_id = ? " + "ORDER BY created_at, event_id", + (str(outcome.result["operation_id"]),), + ).fetchall() + notifications = tuple( + str(row[0]) for row in rows if not str(row[0]).startswith("operation.") + ) + with closing(sqlite3.connect(database)) as live: + saved.backup(live) # the probe leaves no trace in the flow under test + saved.close() + changed = frozenset(name for name in after if after[name] != before.get(name)) + return Repeat( + method=request.method, + accepted=outcome.error is None, + error_message=None if outcome.error is None else str(outcome.error.message), + changed_tables=changed, + added_rows={ + name: len(after[name]) - len(before.get(name, ())) + for name in sorted(changed) + if len(after[name]) != len(before.get(name, ())) + }, + notifications=notifications, + unchanged_reason=unchanged_reason, + cancelled=cancelled, + ) diff --git a/tests/integration/test_stale_running_current_behavior.py b/tests/integration/test_stale_running_current_behavior.py new file mode 100644 index 0000000..0bb53d3 --- /dev/null +++ b/tests/integration/test_stale_running_current_behavior.py @@ -0,0 +1,342 @@ +"""Today's behaviour when the process that was running a general command ends (V-B). + +The change that follows (an owner record, then sealing a request whose owner is confirmed gone) +must change exactly what is pinned here and nothing else. Children are real processes killed for +real, with a synthetic home; no model, login or network is used. + +Observed today: + - a request that was interrupted after its domain change was committed stays RUNNING, and the same + idempotency key returns RUNNING for as long as it is asked (the handler never runs again); + - a request that was interrupted inside its handler stays RUNNING with nothing changed, and only a + new key makes it run; + - the operation row records no process that owned the request, so nothing can tell "owner gone" + from "owner slow" (rows that exist without such information are exactly these rows); + - the domain change and the terminal result are two separate commits for this command; + - the repeat of an applied change under a new key is refused by the revision guard, so nothing is + applied twice. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import time +from pathlib import Path + +import pytest +from tests.integration.storage_coverage_helpers import ( + domain_snapshot, + request, + value, +) +from tests.integration.test_workspace_runtime_exclusivity_baseline import ( + _child_env, # pyright: ignore[reportPrivateUsage] +) + +from thoth.adapters.storage.workspace_owner import WorkspaceInUse +from thoth.apps.runtime import AppRuntime, create_runtime +from thoth.domain.operation import OperationRecord +from thoth.protocol.jsonrpc import JsonRpcRequest + +ROOT = Path(__file__).resolve().parents[2] +PROJECT = "project:stale-running" +_OWNER_PROCESS_COLUMNS = { + "owner_worker_id", + "owner_process_id", + "owner_process_marker", + "owner_host_token", +} + +_CHILD = r""" +import asyncio, os, sys, time +from pathlib import Path + +from thoth.adapters.storage.operations import SqliteOperationStore +from thoth.apps.runtime import create_runtime +from thoth.application.commands.projects import ProjectCommandHandlers +from thoth.protocol.jsonrpc import JsonRpcRequest + + +def build(method, key, values): + return JsonRpcRequest.model_validate({ + "id": key, "method": method, + "params": {"_meta": {"idempotencyKey": key}, "input": values}, + }) + + +async def main(workspace, mode, marker): + runtime = create_runtime(workspace) + created = await runtime.bus.dispatch(build("project/create", "create", { + "project_id": "project:stale-running", "name": "Original", + "cutoff_at": "2026-09-05T00:00:00Z"})) + assert created.error is None, created.error + update = build("project/metadata/update", "rename", { + "project_id": "project:stale-running", "expected_revision": 0, "name": "Renamed"}) + if mode == "after-domain-commit": + def die(*_args, **_kwargs): + Path(marker).write_text(str(os.getpid()), encoding="utf-8") + os._exit(37) + SqliteOperationStore.complete = die + else: + def stuck(self, *_args, **_kwargs): + Path(marker).write_text(str(os.getpid()), encoding="utf-8") + time.sleep(600) + ProjectCommandHandlers._update = stuck + await runtime.bus.dispatch(update) + raise SystemExit("the interruption point was not reached") + + +asyncio.run(main(Path(sys.argv[1]), sys.argv[2], sys.argv[3])) +""" + + +def _interrupted_child(tmp_path: Path, workspace: Path, mode: str) -> int | None: + marker = tmp_path / f"{mode}.marker" + home = tmp_path / "home" + home.mkdir(exist_ok=True) + with (tmp_path / f"{mode}.log").open("wb") as sink: + child = subprocess.Popen( + [sys.executable, "-B", "-c", _CHILD, str(workspace), mode, str(marker)], + cwd=ROOT, + env=_child_env(home), + stdout=sink, + stderr=sink, + ) + deadline = time.monotonic() + 120 + while not marker.is_file() and time.monotonic() < deadline: + assert child.poll() is None, (tmp_path / f"{mode}.log").read_text(errors="replace")[-2000:] + time.sleep(0.05) + assert marker.is_file(), "the child never reached the interruption point" + if mode == "after-domain-commit": + return child.wait(timeout=60) # the child ends itself + child.kill() # a forced kill while the handler is still running + child.wait(timeout=60) + return None + + +def _open(workspace: Path) -> AppRuntime: + """Windows frees a killed process's workspace lock a few milliseconds after it is gone.""" + deadline = time.monotonic() + 10 + while True: + try: + return create_runtime(workspace) + except WorkspaceInUse: + if time.monotonic() > deadline: + raise + time.sleep(0.02) + + +def _rows(runtime: AppRuntime, method: str) -> list[tuple[str, str, str | None]]: + with runtime.ledger.engine.connect() as connection: + found = connection.exec_driver_sql( + "SELECT operation_id, state, completed_at FROM operations WHERE method = ? " + "ORDER BY created_at", + (method,), + ).all() + return [(str(a), str(b), None if c is None else str(c)) for a, b, c in found] + + +def _rename(key: str, expected_revision: int) -> JsonRpcRequest: + return request( + "project/metadata/update", + key, + {"project_id": PROJECT, "expected_revision": expected_revision, "name": "Renamed"}, + ) + + +async def test_a_request_interrupted_after_its_change_was_committed_stays_running( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + assert _interrupted_child(tmp_path, workspace, "after-domain-commit") == 37 + runtime = _open(workspace) + try: + # The change was committed; the terminal result never was. + assert [row[1:] for row in _rows(runtime, "project/metadata/update")] == [("RUNNING", None)] + read = value( + await runtime.bus.query(request("project/read", "after", {"project_id": PROJECT})) + ) + assert read["name"] == "Renamed" and read["revision"] == 1 + + # The same key is answered with RUNNING every time; the handler never runs again. + first = await runtime.bus.dispatch(_rename("rename", 0)) + again = await runtime.bus.dispatch(_rename("rename", 0)) + for replay in (first, again): + assert replay.error is None and replay.result is not None + assert replay.result["state"] == "RUNNING" + assert replay.result["value"] is None + assert replay.result["execution_observation"] == "UNKNOWN" + assert [row[1] for row in _rows(runtime, "project/metadata/update")] == ["RUNNING"] + after_replays = domain_snapshot(runtime.ledger.engine) + + # The repeat of the applied change under a NEW key is refused by the revision guard: + # nothing is applied twice, and the refusal changes no domain row. + repeat = await runtime.bus.dispatch(_rename("rename-again", 0)) + assert repeat.error is not None + assert "expected_revision" in repeat.error.message + assert domain_snapshot(runtime.ledger.engine) == after_replays + assert [row[1] for row in _rows(runtime, "project/metadata/update")] == [ + "RUNNING", + "FAILED", + ] + finally: + runtime.close() + + +async def test_a_request_interrupted_inside_its_handler_stays_running_with_nothing_changed( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + assert _interrupted_child(tmp_path, workspace, "inside-handler") is None + runtime = _open(workspace) + try: + assert [row[1:] for row in _rows(runtime, "project/metadata/update")] == [("RUNNING", None)] + before = value( + await runtime.bus.query(request("project/read", "before", {"project_id": PROJECT})) + ) + assert before["name"] == "Original" and before["revision"] == 0 + + # Asking again with the same key neither runs the change nor closes the request. + for _ in range(2): + replay = await runtime.bus.dispatch(_rename("rename", 0)) + assert replay.error is None and replay.result is not None + assert replay.result["state"] == "RUNNING" and replay.result["value"] is None + unchanged = value( + await runtime.bus.query(request("project/read", "unchanged", {"project_id": PROJECT})) + ) + assert unchanged["name"] == "Original" and unchanged["revision"] == 0 + + # Only a new key runs it, once. + fresh = value(await runtime.bus.dispatch(_rename("rename-new", 0))) + assert fresh["name"] == "Renamed" and fresh["revision"] == 1 + assert [row[1] for row in _rows(runtime, "project/metadata/update")] == [ + "RUNNING", + "SUCCEEDED", + ] + finally: + runtime.close() + + +async def test_the_operation_row_records_no_owning_process(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + assert _interrupted_child(tmp_path, workspace, "after-domain-commit") == 37 + runtime = _open(workspace) + try: + with runtime.ledger.engine.connect() as connection: + columns = { + str(row[1]) + for row in connection.exec_driver_sql("PRAGMA table_info(operations)").all() + } + stuck = ( + connection.exec_driver_sql( + "SELECT * FROM operations WHERE method = 'project/metadata/update'" + ) + .mappings() + .one() + ) + # Only the authenticated actor is recorded, never the process that was running the work, so + # a RUNNING row cannot say whether its owner is gone. D-B adds those columns on purpose. + assert not _OWNER_PROCESS_COLUMNS & columns + assert not _OWNER_PROCESS_COLUMNS & set(OperationRecord.model_fields) + assert stuck["state"] == "RUNNING" + assert json.dumps(dict(stuck), default=str).count(str(workspace)) == 0 + finally: + runtime.close() + + +async def test_the_change_and_the_terminal_result_are_two_separate_commits( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """In-process stand-in for the kill above: the terminal write fails after the change.""" + from thoth.adapters.storage.operations import SqliteOperationStore + + class Died(BaseException): + pass + + def die(*_args: object, **_kwargs: object) -> None: + raise Died + + runtime = create_runtime(tmp_path / "workspace") + try: + value( + await runtime.bus.dispatch( + request( + "project/create", + "create", + { + "project_id": PROJECT, + "name": "Original", + "cutoff_at": "2026-09-05T00:00:00Z", + }, + ) + ) + ) + before = domain_snapshot(runtime.ledger.engine) + monkeypatch.setattr(SqliteOperationStore, "complete", die) + with pytest.raises(Died): + await runtime.bus.dispatch(_rename("rename", 0)) + monkeypatch.undo() + # The change is already in the domain rows, while the operation is still RUNNING: the two + # are separate commits, so an interruption between them leaves exactly this state. + assert domain_snapshot(runtime.ledger.engine) != before + read = value( + await runtime.bus.query(request("project/read", "after", {"project_id": PROJECT})) + ) + assert read["name"] == "Renamed" and read["revision"] == 1 + assert [row[1] for row in _rows(runtime, "project/metadata/update")] == ["RUNNING"] + finally: + runtime.close() + + +_GUARDED_PROJECT_COMMANDS: list[tuple[str, dict[str, object]]] = [ + ("project/metadata/update", {"name": "Renamed"}), + ("project/overlay/update", {"overlay": "interrupted-overlay"}), +] + + +@pytest.mark.parametrize( + ("method", "fields"), + _GUARDED_PROJECT_COMMANDS, + ids=[method for method, _ in _GUARDED_PROJECT_COMMANDS], +) +async def test_an_interrupted_project_change_is_refused_when_asked_again_with_a_new_key( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + method: str, + fields: dict[str, object], +) -> None: + from thoth.adapters.storage.operations import SqliteOperationStore + + class Died(BaseException): + pass + + def die(*_args: object, **_kwargs: object) -> None: + raise Died + + runtime = create_runtime(tmp_path / "workspace") + try: + value( + await runtime.bus.dispatch( + request( + "project/create", + "create", + { + "project_id": PROJECT, + "name": "Original", + "cutoff_at": "2026-09-05T00:00:00Z", + }, + ) + ) + ) + payload: dict[str, object] = {"project_id": PROJECT, "expected_revision": 0, **fields} + monkeypatch.setattr(SqliteOperationStore, "complete", die) + with pytest.raises(Died): + await runtime.bus.dispatch(request(method, "first", payload)) + monkeypatch.undo() + applied = domain_snapshot(runtime.ledger.engine) + repeat = await runtime.bus.dispatch(request(method, "second", payload)) + assert repeat.error is not None and "expected_revision" in repeat.error.message + assert domain_snapshot(runtime.ledger.engine) == applied + finally: + runtime.close() diff --git a/tests/integration/test_state_transition_noop.py b/tests/integration/test_state_transition_noop.py new file mode 100644 index 0000000..ce1ee0a --- /dev/null +++ b/tests/integration/test_state_transition_noop.py @@ -0,0 +1,164 @@ +"""A repeated pause, resume or stop returns the state that is already in place (D-S1). + +Thread (without a research request) and investigation commands: a repeat under a new key returns +the current state with unchanged=true and an unchanged_reason, adds no revision, activity, +checkpoint or audit row and sends no notification. A guard (the checkpoint digest) is still checked +first, and an unsupported stop reason is still refused. Fake runtime only. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from tests.integration.scoped_runtime import create_runtime +from tests.integration.storage_coverage_helpers import request, value + +from thoth.apps.runtime import AppRuntime +from thoth.protocol.notifications import notifications_for + +SCOPE: dict[str, object] = {"project_id": "p", "thread_id": "th"} + + +async def _start(runtime: AppRuntime) -> None: + value( + await runtime.bus.dispatch( + request( + "project/create", + "p", + {"project_id": "p", "name": "X", "cutoff_at": "2026-08-31T00:00:00Z"}, + ) + ) + ) + value( + await runtime.bus.dispatch( + request( + "thread/start", + "t", + {**SCOPE, "problem": "Why?", "scope": {"workstream": "x"}}, + ) + ) + ) + + +async def _written(runtime: AppRuntime) -> tuple[int, int]: + activities = value(await runtime.bus.dispatch(request("thread/activity/list", "a", SCOPE))) + checkpoints = value(await runtime.bus.dispatch(request("thread/checkpoint/list", "c", SCOPE))) + return len(activities["activities"]), len(checkpoints["checkpoints"]) + + +def _operation_events(runtime: AppRuntime, operation_id: str) -> list[str]: + with runtime.ledger.engine.connect() as connection: + rows = connection.exec_driver_sql( + "SELECT event_type FROM events WHERE operation_id = ?", (operation_id,) + ).all() + return [str(row[0]) for row in rows] + + +async def _send(runtime: AppRuntime, method: str, key: str, **fields: object) -> dict[str, Any]: + response = await runtime.bus.dispatch(request(method, key, {**SCOPE, **fields})) + assert response.error is None and response.result is not None, response.error + answered = response.result["value"] + assert isinstance(answered, dict) + return {**answered, "_operation": response.result["operation_id"]} + + +async def test_a_repeated_thread_pause_resume_and_stop_write_nothing(tmp_path: Path) -> None: + runtime = create_runtime(tmp_path / "workspace") + try: + await _start(runtime) + paused = await _send(runtime, "thread/pause", "pause-1") + assert paused["execution_state"] == "PAUSED" and "unchanged" not in paused + digest = paused["checkpoint"]["checkpoint_digest"] + before = await _written(runtime) + + again = await _send(runtime, "thread/pause", "pause-2") + assert again["unchanged"] is True and again["unchanged_reason"] == "THREAD_ALREADY_PAUSED" + assert again["execution_state"] == "PAUSED" and again["revision"] == paused["revision"] + assert again["checkpoint"] == paused["checkpoint"] + assert await _written(runtime) == before + assert not [e for e in _operation_events(runtime, again["_operation"]) if "/" in e] + + # The guard is checked first: a stale digest is still refused while paused. + stale = await runtime.bus.dispatch( + request("thread/resume", "stale", {**SCOPE, "expected_checkpoint_digest": "0" * 64}) + ) + assert stale.error is not None + + resumed = await _send( + runtime, "thread/resume", "resume-1", expected_checkpoint_digest=digest + ) + assert resumed["execution_state"] == "IDLE" and "unchanged" not in resumed + before = await _written(runtime) + # Resuming again with the same digest (still the latest checkpoint) changes nothing. + repeated = await _send( + runtime, "thread/resume", "resume-2", expected_checkpoint_digest=digest + ) + assert repeated["unchanged"] is True + assert repeated["unchanged_reason"] == "THREAD_ALREADY_RESUMED" + assert repeated["revision"] == resumed["revision"] + assert await _written(runtime) == before + assert not [e for e in _operation_events(runtime, repeated["_operation"]) if "/" in e] + + stopped = await _send(runtime, "thread/stop", "stop-1") + assert stopped["lifecycle"] == "STOPPED" and "unchanged" not in stopped + before = await _written(runtime) + stopped_again = await _send(runtime, "thread/stop", "stop-2") + assert stopped_again["unchanged"] is True + assert stopped_again["unchanged_reason"] == "THREAD_ALREADY_STOPPED" + assert stopped_again["revision"] == stopped["revision"] + assert await _written(runtime) == before + assert not [e for e in _operation_events(runtime, stopped_again["_operation"]) if "/" in e] + finally: + runtime.close() + + +async def test_resuming_a_thread_that_was_never_paused_changes_nothing(tmp_path: Path) -> None: + runtime = create_runtime(tmp_path / "workspace") + try: + await _start(runtime) + before = await _written(runtime) + resumed = await _send(runtime, "thread/resume", "resume") + assert resumed["unchanged"] is True + assert resumed["unchanged_reason"] == "THREAD_ALREADY_RESUMED" + assert resumed["revision"] == 0 + assert await _written(runtime) == before + finally: + runtime.close() + + +async def test_a_repeated_investigation_pause_writes_nothing(tmp_path: Path) -> None: + runtime = create_runtime(tmp_path / "workspace") + try: + await _start(runtime) + listed = value(await runtime.bus.dispatch(request("investigation/list", "l", SCOPE))) + scope: dict[str, object] = { + "project_id": "p", + "investigation_id": listed["investigations"][0]["investigation_id"], + } + + async def audit_count() -> int: + read = value( + await runtime.bus.dispatch( + request("investigation/audit/read", "audit-" + str(id(object())), scope) + ) + ) + return len(read["records"]) + + paused = value(await runtime.bus.dispatch(request("investigation/pause", "p1", scope))) + assert paused["investigation"]["execution_state"] == "PAUSED" + before = await audit_count() + again = value(await runtime.bus.dispatch(request("investigation/pause", "p2", scope))) + assert again["unchanged"] is True + assert again["unchanged_reason"] == "INVESTIGATION_ALREADY_PAUSED" + assert again["investigation"] == paused["investigation"] + assert await audit_count() == before + finally: + runtime.close() + + +def test_an_unchanged_result_announces_nothing_and_a_changed_one_still_does() -> None: + assert notifications_for("thread/resume", {"unchanged": True}) == () + assert notifications_for("thread/stop", {"unchanged": True, "cancelled": False}) == () + assert "thread/resumed" in notifications_for("thread/resume", {}) + assert "thread/resumed" in notifications_for("thread/resume", {"unchanged": False}) diff --git a/tests/integration/test_state_transition_repeat_current_behavior.py b/tests/integration/test_state_transition_repeat_current_behavior.py new file mode 100644 index 0000000..06d7a21 --- /dev/null +++ b/tests/integration/test_state_transition_repeat_current_behavior.py @@ -0,0 +1,408 @@ +"""How a state-changing command answers a repeat under a new key (V-S; D-S1 and D-S2 changed it). + +Nine commands change a state: thread pause/resume/stop, investigation pause/stop, closure +decide/reopen, revision baseline decide and action authorization decide. For all of them an +"already in that state" repeat now returns the current state with unchanged=true, writes no domain +row and sends no notification (V-S pinned the opposite; D-S1 changed the thread and investigation +commands, D-S2 the rest). Also pinned here: + - an investigation that already ended keeps its verdict and result whatever stop reason comes + later, and is not paused; + - authorization: a repeat of the same role's vote while the approval is PENDING is unchanged; once + the decision is final a repeat is still refused ("authorization is not pending"); + - the research continuation path (thread pause/resume/stop while a research request exists): + a repeated pause is answered unchanged, "already terminal" is answered once the request has + finished, a repeated stop is answered unchanged without cancelling again (and still cancels a + request a cut-off first stop left running), and a second resume while the first is still + running is answered unchanged and starts no second research (D-S3). +The first run of each command, the same-key replay and the guard refusals are not pinned here (the +existing flow tests cover them and must keep passing unchanged). The flow tests are run as they are +and the repeat is sent from a wrapper around the bus (see repeat_probe.py). Fake models only. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from pathlib import Path +from typing import Any + +import pytest +from tests.integration.repeat_probe import RepeatProbe +from tests.integration.scoped_runtime import create_runtime +from tests.integration.storage_coverage_helpers import domain_snapshot, request, value +from tests.integration.test_action_full_rpc import ( + test_action_decision_plan_policy_and_content_bound_authorization as action_flow, +) +from tests.integration.test_investigation_full_rpc import ( + test_investigation_plan_checkpoint_audit_and_stop_reducer as investigation_flow, +) +from tests.integration.test_receipt_closure_export_full_rpc import ( + test_receipt_closure_and_local_export_without_external_release as closure_flow, +) +from tests.integration.test_research_request_v2 import ControlledResearchModel, setup +from tests.integration.test_revision_full_rpc import ( + test_revision_atomic_changeset_merge_restore_and_baseline as baseline_flow, +) +from tests.integration.test_thread_full_catalog_rpc import ( + test_thread_list_activity_checkpoint_steer_fork_and_stop as thread_flow, +) + +from thoth.apps.runtime import AppRuntime + +Flow = Callable[[Path], Awaitable[None]] + +# method -> why the repeat is answered unchanged +_UNCHANGED: list[tuple[str, Flow, dict[str, str]]] = [ + ( + "thread", + thread_flow, + { + "thread/pause": "THREAD_ALREADY_PAUSED", + "thread/resume": "THREAD_ALREADY_RESUMED", + "thread/stop": "THREAD_ALREADY_STOPPED", + }, + ), + ( + "investigation", + investigation_flow, + { + "investigation/pause": "INVESTIGATION_ALREADY_PAUSED", + "investigation/stop": "INVESTIGATION_ALREADY_ENDED", + }, + ), + ( + "closure", + closure_flow, + { + "closure/decide": "CLOSURE_ALREADY_DECIDED", + "closure/reopen": "CLOSURE_ALREADY_REOPENED", + }, + ), + ("baseline", baseline_flow, {"revision/baseline/decide": "BASELINE_ALREADY_DECIDED"}), + ( + "authorization", + action_flow, + {"action/authorization/decide": "AUTHORIZATION_VOTE_ALREADY_RECORDED"}, + ), +] + + +@pytest.mark.parametrize( + ("name", "flow", "reasons"), _UNCHANGED, ids=[name for name, _flow, _reasons in _UNCHANGED] +) +async def test_a_repeat_of_the_same_state_returns_it_and_writes_nothing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + name: str, + flow: Flow, + reasons: dict[str, str], +) -> None: + probe = RepeatProbe(frozenset(reasons)) + probe.install(monkeypatch) + await flow(tmp_path) + for method, reason in reasons.items(): + repeats = probe.results[method] + assert repeats, (name, method) + for repeat in repeats: + if not repeat.accepted: + # Only a final authorization decision is still refused, and refused without writing. + assert method == "action/authorization/decide", (method, repeat) + assert repeat.error_message == "authorization is not pending", repeat + else: + assert repeat.unchanged_reason == reason, (method, repeat) + assert not repeat.changed_tables and not repeat.added_rows, (method, repeat) + assert repeat.notifications == (), (method, repeat) + if name == "authorization": + refused = [r for r in probe.results["action/authorization/decide"] if not r.accepted] + assert refused, "a final decision must still refuse a repeat" + + +async def test_a_different_decision_or_scope_is_still_recorded( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Only the same decision (or the same reopening scope) is answered unchanged.""" + + def change(method: str, params: dict[str, Any]) -> None: + if method == "closure/decide": + params["decision"] = "REJECT" + else: + params["child_scope_ref"] = "scope:another-child" + + probe = RepeatProbe(frozenset({"closure/decide", "closure/reopen"}), change=change) + probe.install(monkeypatch) + await closure_flow(tmp_path) + for method in ("closure/decide", "closure/reopen"): + for repeat in probe.results[method]: + assert repeat.accepted and repeat.unchanged_reason is None, (method, repeat) + assert repeat.changed_tables == {"control_records"}, (method, repeat) + + +async def test_a_later_stop_keeps_the_verdict_of_an_investigation_that_already_ended( + tmp_path: Path, +) -> None: + runtime = create_runtime(tmp_path / "workspace") + try: + value( + await runtime.bus.dispatch( + request( + "project/create", + "p", + {"project_id": "p", "name": "X", "cutoff_at": "2026-08-31T00:00:00Z"}, + ) + ) + ) + value( + await runtime.bus.dispatch( + request( + "thread/start", + "t", + { + "project_id": "p", + "thread_id": "th", + "problem": "Why?", + "scope": {"workstream": "x"}, + }, + ) + ) + ) + listed = value( + await runtime.bus.dispatch( + request("investigation/list", "l", {"project_id": "p", "thread_id": "th"}) + ) + ) + investigation_id = listed["investigations"][0]["investigation_id"] + + async def send(method: str, key: str, **fields: object) -> dict[str, Any]: + return value( + await runtime.bus.dispatch( + request( + method, + key, + {"project_id": "p", "investigation_id": investigation_id, **fields}, + ) + ) + ) + + first = await send("investigation/stop", "stop-1", reason="SUFFICIENT") + converged = first["investigation"] + assert converged["domain_state"] == "CONVERGED" + assert converged["result"]["stop_reason"] == "SUFFICIENT" + assert "unchanged" not in first + # A later stop (any supported reason) answers with the finished state and changes nothing. + later = await send("investigation/stop", "stop-2", reason="USER_STOPPED") + assert later["unchanged"] is True + assert later["unchanged_reason"] == "INVESTIGATION_ALREADY_ENDED" + assert later["investigation"] == converged + # An unsupported reason is still refused, as before. + refused = await runtime.bus.dispatch( + request( + "investigation/stop", + "stop-3", + { + "project_id": "p", + "investigation_id": investigation_id, + "reason": "NOT_A_REASON", + }, + ) + ) + assert refused.error is not None + # An ended investigation is not paused either. + paused = await send("investigation/pause", "pause") + assert paused["unchanged"] is True + assert paused["investigation"] == converged + assert paused["investigation"]["execution_state"] == "IDLE" + finally: + runtime.close() + + +async def _thread_state(runtime: AppRuntime, scope: dict[str, object]) -> tuple[Any, Any]: + read = value( + await runtime.bus.dispatch(request("thread/read", "read-" + str(id(object())), scope)) + ) + return read["execution_state"], read["revision"] + + +async def test_the_research_path_answers_a_repeated_pause_and_stop_unchanged( + tmp_path: Path, +) -> None: + model = ControlledResearchModel(wait=True) + runtime = await setup(tmp_path, model, source=False) + try: + accepted = value( + await runtime.bus.dispatch( + request( + "thread/start", + "start", + {"project_id": "p", "problem": "Pause then resume", "contract_version": 2}, + ) + ) + ) + await asyncio.wait_for(model.started.wait(), 8) + scope: dict[str, object] = {"project_id": "p", "thread_id": accepted["thread_id"]} + + async def send(method: str, key: str) -> dict[str, Any]: + return value(await runtime.bus.dispatch(request(method, key, scope))) + + # While the research still runs, a second pause is answered unchanged and writes nothing. + first = await send("thread/pause", "pause-1") + _state, revision = await _thread_state(runtime, scope) + assert first["pause_accepted"] is True and first["execution_state"] == "PAUSE_PENDING" + second = await send("thread/pause", "pause-2") + assert second["pause_accepted"] is True and second["execution_state"] == "PAUSE_PENDING" + assert second["unchanged"] is True and second["unchanged_reason"] == "THREAD_ALREADY_PAUSED" + assert await _thread_state(runtime, scope) == ("PAUSE_PENDING", revision) + model.release.set() + await asyncio.wait_for(runtime.bus.drain(), 12) + state, revision = await _thread_state(runtime, scope) + assert state == "PAUSED" + # Already paused: a pause is answered unchanged and the revision does not move. + third = await send("thread/pause", "pause-3") + assert third["pause_accepted"] is True and third["execution_state"] == "PAUSED" + assert third["unchanged"] is True and third["unchanged_reason"] == "THREAD_ALREADY_PAUSED" + assert await _thread_state(runtime, scope) == ("PAUSED", revision) + + await send("thread/resume", "resume-1") + await asyncio.wait_for(runtime.bus.drain(), 12) + operation = runtime.bus.read_operation(str(accepted["operation_id"])) + assert operation is not None and operation.state.value == "SUCCEEDED" + finished = await _thread_state(runtime, scope) + # Once the request has finished, pause and resume answer "already terminal", write nothing. + assert await send("thread/resume", "resume-2") == { + "resume_allowed": False, + "reason": "OPERATION_ALREADY_TERMINAL", + } + assert await send("thread/pause", "pause-4") == { + "pause_accepted": False, + "reason": "OPERATION_ALREADY_TERMINAL", + } + assert await _thread_state(runtime, scope) == finished + # The first stop cancels the request; a repeated stop answers unchanged, cancels nothing + # new (cancelled=False means "this request cancelled nothing") and writes nothing. + stopped = await send("thread/stop", "stop-1") + assert stopped["lifecycle"] == "STOPPED" and stopped["cancelled"] is True + assert "unchanged" not in stopped + after_stop = await _thread_state(runtime, scope) + assert after_stop == (finished[0], finished[1] + 1) + again = await send("thread/stop", "stop-2") + assert again["lifecycle"] == "STOPPED" and again["cancelled"] is False + assert again["unchanged"] is True and again["unchanged_reason"] == "THREAD_ALREADY_STOPPED" + assert await _thread_state(runtime, scope) == after_stop + finally: + runtime.close() + + +async def _research_calls(tmp_path: Path, *, resumes: int) -> tuple[int, str, Any, dict[str, Any]]: + """Pause a running research request, then resume it `resumes` times while it runs again.""" + model = ControlledResearchModel(wait=True) + runtime = await setup(tmp_path, model, source=False) + try: + accepted = value( + await runtime.bus.dispatch( + request( + "thread/start", + "start", + {"project_id": "p", "problem": "Resume twice", "contract_version": 2}, + ) + ) + ) + await asyncio.wait_for(model.started.wait(), 8) + scope: dict[str, object] = {"project_id": "p", "thread_id": accepted["thread_id"]} + value(await runtime.bus.dispatch(request("thread/pause", "pause", scope))) + model.release.set() + await asyncio.wait_for(runtime.bus.drain(), 12) + model.release.clear() # the resumed run blocks again, so the next resume meets it running + last: dict[str, Any] = {} + for index in range(resumes): + last = value( + await runtime.bus.dispatch(request("thread/resume", f"resume-{index}", scope)) + ) + await asyncio.sleep(0.5) + model.release.set() + await asyncio.wait_for(runtime.bus.drain(), 12) + operation = runtime.bus.read_operation(str(accepted["operation_id"])) + assert operation is not None + state = await _thread_state(runtime, scope) + return len(model.calls), operation.state.value, state, last + finally: + runtime.close() + + +async def test_a_second_resume_while_the_first_is_running_starts_no_second_research( + tmp_path: Path, +) -> None: + once_calls, once_state, once_thread, once_last = await _research_calls( + tmp_path / "once", resumes=1 + ) + twice_calls, twice_state, twice_thread, twice_last = await _research_calls( + tmp_path / "twice", resumes=2 + ) + assert once_state == twice_state == "SUCCEEDED" + # No duplicate model work, and the second resume (met while the first still runs) is answered + # unchanged: it writes nothing and dispatches nothing, so the thread ends as after one resume. + assert twice_calls == once_calls + assert "unchanged" not in once_last + assert twice_last["unchanged"] is True + assert twice_last["unchanged_reason"] == "THREAD_ALREADY_RESUMED" + assert twice_last["resume_allowed"] is True + assert twice_thread == once_thread + + +async def test_a_second_stop_finishes_the_cancel_that_a_cut_off_first_stop_did_not_reach( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from thoth.adapters.storage.operations import SqliteOperationStore + + class Died(BaseException): + pass + + def die(*_args: object, **_kwargs: object) -> None: + raise Died + + model = ControlledResearchModel(wait=True) + runtime = await setup(tmp_path, model, source=False) + try: + accepted = value( + await runtime.bus.dispatch( + request( + "thread/start", + "start", + {"project_id": "p", "problem": "Stop twice", "contract_version": 2}, + ) + ) + ) + await asyncio.wait_for(model.started.wait(), 8) + scope: dict[str, object] = {"project_id": "p", "thread_id": accepted["thread_id"]} + operation_id = str(accepted["operation_id"]) + + def operation_state() -> str: + operation = runtime.bus.read_operation(operation_id) + assert operation is not None + return operation.state.value + + # The first stop commits the thread as STOPPED, then is cut off before it can cancel. + with monkeypatch.context() as patch: + patch.setattr(SqliteOperationStore, "cancel", die) + with pytest.raises(Died): + await runtime.bus.dispatch(request("thread/stop", "stop-1", scope)) + read = value(await runtime.bus.dispatch(request("thread/read", "read-1", scope))) + assert read["lifecycle"] == "STOPPED" + assert operation_state() == "RUNNING" + stopped_revision = read["revision"] + + # A second stop does not skip the cancel just because the thread is already stopped. + second = value(await runtime.bus.dispatch(request("thread/stop", "stop-2", scope))) + assert second["cancelled"] is True and second["cancel_operation_id"] == operation_id + assert "unchanged" not in second # announced as a normal stop, the cancel is not hidden + await asyncio.wait_for(runtime.bus.drain(), 12) + assert operation_state() == "CANCELLED" + read = value(await runtime.bus.dispatch(request("thread/read", "read-2", scope))) + assert read["revision"] == stopped_revision # the thread itself was not written again + + # Once the request is cancelled, a further stop writes nothing and cancels nothing. + before = domain_snapshot(runtime.ledger.engine) + third = value(await runtime.bus.dispatch(request("thread/stop", "stop-3", scope))) + assert third["unchanged"] is True and third["cancelled"] is False + assert domain_snapshot(runtime.ledger.engine) == before + assert operation_state() == "CANCELLED" + finally: + runtime.close() diff --git a/tests/integration/test_workspace_runtime_exclusivity_baseline.py b/tests/integration/test_workspace_runtime_exclusivity_baseline.py new file mode 100644 index 0000000..95cb3a7 --- /dev/null +++ b/tests/integration/test_workspace_runtime_exclusivity_baseline.py @@ -0,0 +1,224 @@ +"""Who may open a workspace: the single-runtime rule and what stays as it was (V-A, then D-A). + + - a second PROCESS is refused when another process already has the workspace open + (before D-A this test pinned that it was allowed); + - several runtimes inside ONE process share a workspace; + - the local server opens its runtime on the first request, not at start-up; + - the commands that do not use the database leave no database behind, while `events` and + `run` open a runtime. +No model, login, network or real credential is used; child processes get a synthetic home. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path + +import httpx +import pytest +from typer.testing import CliRunner + +from thoth.adapters.http.app import create_app +from thoth.adapters.storage.workspace_owner import WorkspaceInUse +from thoth.apps.runtime import AppRuntime, create_runtime +from thoth.cli import app as cli_app + +ROOT = Path(__file__).resolve().parents[2] + +_HOLDER = r""" +import os +import sys +from pathlib import Path +from thoth.apps.runtime import create_runtime + +workspace, marker = Path(sys.argv[1]), Path(sys.argv[2]) +runtime = create_runtime(workspace) +marker.write_text(str(os.getpid()), encoding="utf-8") +sys.stdin.readline() +runtime.close() +""" + + +def _child_env(home: Path) -> dict[str, str]: + return { + **os.environ, + "PYTHONDONTWRITEBYTECODE": "1", + "HOME": str(home), + "USERPROFILE": str(home), + "LOCALAPPDATA": str(home / "local"), + "APPDATA": str(home / "roaming"), + "XDG_DATA_HOME": str(home / "xdg"), + "CODEX_HOME": str(home / "codex"), + } + + +def _projects(runtime: AppRuntime) -> int: + with runtime.ledger.engine.connect() as connection: + return int(connection.exec_driver_sql("SELECT count(*) FROM projects").scalar()) + + +def test_a_second_process_cannot_open_a_workspace_another_process_has_open(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + marker = tmp_path / "holder-open.marker" + home = tmp_path / "home" + home.mkdir() + log = tmp_path / "holder.log" + # The child's migration log is long; a full pipe would block it, so write it to a file. + with log.open("wb") as sink: + holder = subprocess.Popen( + [sys.executable, "-B", "-c", _HOLDER, str(workspace), str(marker)], + env=_child_env(home), + stdin=subprocess.PIPE, + stdout=sink, + stderr=sink, + text=True, + ) + try: + deadline = time.monotonic() + 60 + while not marker.is_file() and time.monotonic() < deadline: + assert holder.poll() is None, log.read_text(encoding="utf-8", errors="replace")[-2000:] + time.sleep(0.05) + assert marker.is_file() + with pytest.raises(WorkspaceInUse) as refused: + create_runtime(workspace) + assert refused.value.holder is not None + # On Windows the launched process can be a launcher, so compare with the child's own id. + assert refused.value.holder.process_id == int(marker.read_text(encoding="utf-8")) + assert holder.poll() is None + finally: + if holder.poll() is None and holder.stdin is not None: + holder.stdin.write("\n") + holder.stdin.flush() + try: + holder.wait(timeout=30) + except subprocess.TimeoutExpired: + holder.kill() + holder.wait(timeout=10) + assert holder.returncode == 0, log.read_text(encoding="utf-8", errors="replace")[-2000:] + + +def test_runtimes_in_one_process_share_a_workspace_until_each_is_closed(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + first = create_runtime(workspace) + second = create_runtime(workspace) + try: + assert _projects(first) == _projects(second) == 0 + assert first.bus.registered_methods() == second.bus.registered_methods() + first.close() + assert _projects(second) == 0 + finally: + second.close() + reopened = create_runtime(workspace) + try: + assert _projects(reopened) == 0 + finally: + reopened.close() + + +async def test_the_local_server_opens_its_runtime_on_the_first_request( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspace = tmp_path / "server-workspace" + monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) + monkeypatch.delenv("THOTH_DEPLOYMENT_MODE", raising=False) + application = create_app() + transport = httpx.ASGITransport(app=application) + try: + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + assert (await client.get("/healthz")).status_code == 200 + assert application.state.runtime is None + assert not (workspace / "db").exists() + response = await client.post( + "/rpc", + json={ + "jsonrpc": "2.0", + "id": "ready-1", + "method": "workspace/ready", + "params": {"_meta": {"idempotencyKey": "baseline-ready"}, "input": {}}, + }, + ) + assert response.status_code == 200 + assert application.state.runtime is not None + assert (workspace / "db" / "thoth.sqlite3").is_file() + finally: + if application.state.runtime is not None: + application.state.runtime.close() + + +def _git_repository(path: Path) -> Path: + path.mkdir() + identity = ["-c", "user.name=synthetic", "-c", "user.email=synthetic@example.invalid"] + for arguments in ( + ["init", "-q"], + [*identity, "commit", "--allow-empty", "-q", "-m", "synthetic"], + ): + subprocess.run(["git", "-C", str(path), *arguments], check=True, capture_output=True) + return path + + +def test_commands_that_do_not_use_the_database_leave_no_database(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + source = tmp_path / "note.txt" + source.write_text("synthetic source", encoding="utf-8") + runner = CliRunner() + shown = runner.invoke(cli_app, ["workspace", "--workspace", str(workspace)]) + assert shown.exit_code == 0 + assert json.loads(shown.stdout)["workspace"] == str(workspace.resolve()) + runner.invoke(cli_app, ["doctor", "--workspace", str(workspace), "--json"]) + staged = runner.invoke( + cli_app, ["source-stage", "--source", str(source), "--workspace", str(workspace)] + ) + assert staged.exit_code == 0 + snapshot = runner.invoke( + cli_app, + [ + "git-snapshot", + "--repository", + str(_git_repository(tmp_path / "repository")), + "--workspace", + str(workspace), + ], + ) + assert snapshot.exit_code == 0 + assert (workspace / "inbox").is_dir() + assert not (workspace / "db").exists() + + +def test_events_and_run_open_a_runtime_and_create_the_database(tmp_path: Path) -> None: + runner = CliRunner() + events_workspace = tmp_path / "events" + runner.invoke( + cli_app, + [ + "events", + "--project-id", + "project:absent", + "--operation-id", + "operation:absent", + "--workspace", + str(events_workspace), + ], + ) + assert (events_workspace / "db" / "thoth.sqlite3").is_file() + request = tmp_path / "request.json" + request.write_text( + json.dumps( + { + "jsonrpc": "2.0", + "id": "ready-1", + "method": "workspace/ready", + "params": {"_meta": {"idempotencyKey": "baseline-run"}, "input": {}}, + } + ), + encoding="utf-8", + ) + run_workspace = tmp_path / "run" + result = runner.invoke( + cli_app, ["run", "--request", str(request), "--workspace", str(run_workspace)] + ) + assert result.exit_code == 0 + assert (run_workspace / "db" / "thoth.sqlite3").is_file() diff --git a/tests/integration/test_workspace_runtime_lock.py b/tests/integration/test_workspace_runtime_lock.py new file mode 100644 index 0000000..88b1a8c --- /dev/null +++ b/tests/integration/test_workspace_runtime_lock.py @@ -0,0 +1,364 @@ +"""One THOTH process per workspace (D-A): refusal, release, and what the refusal never writes. + +Children are real processes with a synthetic home; no model, login or network is used. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import sqlite3 +import subprocess +import sys +import time +from collections.abc import Generator +from contextlib import contextmanager +from pathlib import Path + +import httpx +import pytest +from tests.integration.test_workspace_runtime_exclusivity_baseline import ( + _HOLDER, # pyright: ignore[reportPrivateUsage] + _child_env, # pyright: ignore[reportPrivateUsage] +) +from typer.testing import CliRunner, Result + +from thoth.adapters.http.app import create_app +from thoth.adapters.storage.bundle import SqliteStoreFactory +from thoth.adapters.storage.workspace_owner import ( + LOCK_FILE, + WorkspaceHolder, + WorkspaceInUse, + acquire_workspace_owner, +) +from thoth.apps.runtime import AppRuntime, create_runtime +from thoth.cli import app as cli_app + +_LOCK_HOLDER = r""" +import os, sys +from pathlib import Path +from thoth.adapters.storage.workspace_owner import acquire_workspace_owner + +workspace, marker = Path(sys.argv[1]), Path(sys.argv[2]) +ownership = acquire_workspace_owner(workspace, command="thoth serve") +marker.write_text(str(os.getpid()), encoding="utf-8") +sys.stdin.readline() +ownership.release() +""" + +_TRY_ACQUIRE = r""" +import sys +from pathlib import Path +from thoth.adapters.storage.workspace_owner import WorkspaceInUse, acquire_workspace_owner + +try: + acquire_workspace_owner(Path(sys.argv[1])).release() + print("OK") +except WorkspaceInUse: + print("IN_USE") +""" + + +@contextmanager +def _holder(tmp_path: Path, workspace: Path, script: str, *, kill: bool = False) -> Generator[int]: + marker = tmp_path / f"holder-{abs(hash(script)) % 10**6}.marker" + home = tmp_path / "home" + home.mkdir(exist_ok=True) + with (tmp_path / "holder.log").open("wb") as sink: + child = subprocess.Popen( + [sys.executable, "-B", "-c", script, str(workspace), str(marker)], + env=_child_env(home), + stdin=subprocess.PIPE, + stdout=sink, + stderr=sink, + text=True, + ) + try: + deadline = time.monotonic() + 90 + while not marker.is_file() and time.monotonic() < deadline: + assert child.poll() is None, (tmp_path / "holder.log").read_text(errors="replace")[ + -1500: + ] + time.sleep(0.05) + assert marker.is_file() + yield int(marker.read_text(encoding="utf-8")) + finally: + if kill and child.poll() is None: + child.kill() + elif child.poll() is None and child.stdin is not None: + child.stdin.write("\n") + child.stdin.flush() + try: + child.wait(timeout=30) + except subprocess.TimeoutExpired: + child.kill() + child.wait(timeout=10) + + +def _can_acquire(tmp_path: Path, workspace: Path) -> bool: + home = tmp_path / "home" + home.mkdir(exist_ok=True) + completed = subprocess.run( + [sys.executable, "-B", "-c", _TRY_ACQUIRE, str(workspace)], + env=_child_env(home), + capture_output=True, + text=True, + timeout=90, + check=False, + ) + assert completed.stdout.strip() in {"OK", "IN_USE"}, completed.stderr[-1500:] + return completed.stdout.strip() == "OK" + + +def test_refusal_names_the_other_process_in_korean_and_english(tmp_path: Path) -> None: + holder = WorkspaceHolder( + process_id=4242, started_at="2026-10-17 09:12 UTC", command="thoth serve" + ) + error = WorkspaceInUse(tmp_path / "ws", holder) + assert "이미 다른 THOTH가 사용 중" in error.message_ko + assert "PID 4242" in error.message_ko and "thoth serve" in error.message_ko + assert "먼저 종료한 뒤 다시 실행하세요" in error.message_ko + assert "already in use by another THOTH" in error.message_en + assert "PID 4242" in error.message_en and "Close that one first" in error.message_en + assert str(tmp_path / "ws") in error.message_ko and str(tmp_path / "ws") in error.message_en + bare = WorkspaceInUse(tmp_path / "ws", None) + assert "PID" not in bare.message_ko and "PID" not in bare.message_en + assert "이미 다른 THOTH가 사용 중입니다." in bare.message_ko + + +def test_one_process_shares_the_lock_by_reference_count(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + first = acquire_workspace_owner(workspace) + second = acquire_workspace_owner(workspace) + try: + assert not _can_acquire(tmp_path, workspace) + first.release() + first.release() # releasing twice must not give away the second reference + assert not _can_acquire(tmp_path, workspace) + finally: + second.release() + assert _can_acquire(tmp_path, workspace) + assert (workspace / LOCK_FILE).stat().st_size == 1 # the hint is cleared on a clean release + + +def _hint_bytes(workspace: Path) -> bytes: + """Read the lock file from byte 1: byte 0 is the lock itself and cannot be read while held.""" + with (workspace / LOCK_FILE).open("rb") as handle: + handle.seek(1) + return handle.read() + + +def _create_runtime_once_the_os_frees_the_lock(workspace: Path) -> AppRuntime: + """Windows releases a killed process's lock a few milliseconds after the process is gone.""" + deadline = time.monotonic() + 10 + while True: + try: + return create_runtime(workspace) + except WorkspaceInUse: + if time.monotonic() > deadline: + raise + time.sleep(0.02) + + +def _tables(database: Path) -> dict[str, int]: + with sqlite3.connect(f"file:{database.as_posix()}?mode=ro", uri=True) as connection: + names = [ + str(row[0]) + for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'") + ] + return { + name: int(connection.execute(f'SELECT count(*) FROM "{name}"').fetchone()[0]) + for name in names + } + + +def _file_hashes(root: Path) -> dict[str, str]: + return { + path.relative_to(root).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest() + for path in sorted(root.rglob("*")) + if path.is_file() and path.name != LOCK_FILE + } + + +def test_a_refused_second_process_writes_nothing_to_the_first_ones_workspace( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + with _holder(tmp_path, workspace, _HOLDER) as holder_pid: + database = workspace / "db" / "thoth.sqlite3" + watcher = sqlite3.connect(f"file:{database.as_posix()}?mode=ro", uri=True) + try: + version = watcher.execute("PRAGMA data_version").fetchone()[0] + tables, files = _tables(database), _file_hashes(workspace) + with pytest.raises(WorkspaceInUse) as refused: + create_runtime(workspace) + assert refused.value.holder is not None + assert refused.value.holder.process_id == holder_pid + assert watcher.execute("PRAGMA data_version").fetchone()[0] == version + assert _tables(database) == tables + assert ( + _file_hashes(workspace) == files + ) # database, WAL, objects, diagnostics: unchanged + finally: + watcher.close() + # The process id and path are only in the lock file, which is the one place they may be. + hint = json.loads(_hint_bytes(workspace).decode("utf-8")) + assert hint["pid"] == holder_pid + + +def test_a_killed_holder_frees_the_workspace_and_its_stale_lock_file_does_not_block( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + with ( + _holder(tmp_path, workspace, _LOCK_HOLDER, kill=True) as holder_pid, + pytest.raises(WorkspaceInUse), + ): + acquire_workspace_owner(workspace) + stale = _hint_bytes(workspace) + assert str(holder_pid).encode() in stale # the killed process never cleared its hint + runtime = _create_runtime_once_the_os_frees_the_lock(workspace) + try: + assert not _can_acquire(tmp_path, workspace) + assert str(os.getpid()).encode() in _hint_bytes(workspace) + finally: + runtime.close() + assert _can_acquire(tmp_path, workspace) + + +def test_closing_a_runtime_releases_the_workspace(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + runtime = create_runtime(workspace) + assert not _can_acquire(tmp_path, workspace) + runtime.close() + assert _can_acquire(tmp_path, workspace) + + +def test_a_failed_start_releases_the_workspace( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspace = tmp_path / "workspace" + + def broken(_database: Path) -> None: + raise RuntimeError("synthetic migration failure") + + monkeypatch.setattr("thoth.adapters.storage.bundle.migrate_sqlite_database", broken) + with pytest.raises(RuntimeError, match="synthetic migration failure"): + create_runtime(workspace) + monkeypatch.undo() + assert _can_acquire(tmp_path, workspace) + create_runtime(workspace).close() + + +def test_the_hosted_review_mode_is_not_locked( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspace = tmp_path / "workspace" + with _holder(tmp_path, workspace, _LOCK_HOLDER): + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "HOSTED_REVIEW") + stores = SqliteStoreFactory().open(workspace) + stores.close() + + +def _blocked_cli(tmp_path: Path, arguments: list[str], monkeypatch: pytest.MonkeyPatch) -> Result: + import uvicorn + + def never_listens(*_args: object, **_kwargs: object) -> None: + raise AssertionError("a refused server must not start listening") + + monkeypatch.setattr(uvicorn, "run", never_listens) + request = tmp_path / "request.json" + request.write_text( + json.dumps( + { + "jsonrpc": "2.0", + "id": "ready-1", + "method": "workspace/ready", + "params": {"_meta": {"idempotencyKey": "locked-run"}, "input": {}}, + } + ), + encoding="utf-8", + ) + workspace = tmp_path / "workspace" + expanded = [ + part.replace("", str(workspace)).replace("", str(request)) for part in arguments + ] + with _holder(tmp_path, workspace, _LOCK_HOLDER): + return CliRunner().invoke(cli_app, expanded, input="/quit\n") + + +@pytest.mark.parametrize( + "arguments", + [ + ["serve", "--workspace", ""], + ["desktop", "--workspace", ""], + ["events", "--project-id", "p", "--operation-id", "o", "--workspace", ""], + ["run", "--request", "", "--workspace", ""], + ["rpc", "--workspace", ""], + ["--workspace", ""], + ], + ids=["serve", "desktop", "events", "run", "rpc", "shell"], +) +def test_every_command_that_opens_the_database_is_refused_with_code_3( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, arguments: list[str] +) -> None: + result = _blocked_cli(tmp_path, arguments, monkeypatch) + assert result.exit_code == 3, result.output + shown = result.output + assert "이미 다른 THOTH가 사용 중입니다" in shown + assert "already in use by another THOTH" in shown + assert not (tmp_path / "workspace" / "db").exists() # refused before any migration started + + +def test_serve_owns_the_workspace_before_it_listens_and_lets_go_after( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import uvicorn + + workspace = tmp_path / "workspace" + observed: list[bool] = [] + + def listens(*_args: object, **_kwargs: object) -> None: + observed.append(_can_acquire(tmp_path, workspace)) # owned before the first request + runtime = create_runtime(workspace) # the first request, in the same process + runtime.close() + observed.append(_can_acquire(tmp_path, workspace)) # serve still owns it + + monkeypatch.setattr(uvicorn, "run", listens) + result = CliRunner().invoke(cli_app, ["serve", "--workspace", str(workspace)]) + assert result.exit_code == 0, result.output + assert observed == [False, False] + assert _can_acquire(tmp_path, workspace) + + +async def test_a_server_not_started_by_the_cli_answers_503_until_the_workspace_is_free( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspace = tmp_path / "workspace" + monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) + monkeypatch.delenv("THOTH_DEPLOYMENT_MODE", raising=False) + application = create_app() + body = { + "jsonrpc": "2.0", + "id": "ready-1", + "method": "workspace/ready", + "params": {"_meta": {"idempotencyKey": "locked-http"}, "input": {}}, + } + transport = httpx.ASGITransport(app=application) + try: + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + with _holder(tmp_path, workspace, _LOCK_HOLDER): + refused = await client.post("/rpc", json=body) + assert refused.status_code == 503 + detail = refused.json()["detail"] + assert detail["code"] == "WORKSPACE_IN_USE" + assert "이미 다른 THOTH가 사용 중입니다" in detail["message_ko"] + assert "already in use by another THOTH" in detail["message_en"] + assert application.state.runtime is None + allowed = await client.post("/rpc", json=body) + assert allowed.status_code == 200 + assert application.state.runtime is not None + finally: + if application.state.runtime is not None: + application.state.runtime.close() diff --git a/tests/unit/application/test_research_controls_repeat.py b/tests/unit/application/test_research_controls_repeat.py new file mode 100644 index 0000000..1a426c0 --- /dev/null +++ b/tests/unit/application/test_research_controls_repeat.py @@ -0,0 +1,151 @@ +"""Research pause/resume: when a repeat is answered unchanged and when it is not (D-S3). + +A repeated pause is unchanged only when the thread AND the attempt are already in the pause that +would be written. A resume is unchanged only when the request is already running under a LIVE +worker (attempt RUNNING, lease HELD and unexpired, holder known to be alive). A dead, unknown or +expired holder, a released or missing lease, or a paused attempt still takes the recovery path: +the thread is written and the attempt is dispatched again. Stand-in host; no storage or model. +""" + +# pyright: reportUnknownLambdaType=false +from __future__ import annotations + +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from typing import Any + +import pytest + +from thoth.application.services.research_controls import ResearchControls +from thoth.domain.enums import OperationState, ThreadExecutionState + +NOW = datetime(2026, 10, 18, 12, 0, tzinfo=UTC) +SCOPE: dict[str, Any] = {"project_id": "p", "thread_id": "th"} + + +class _Copy(SimpleNamespace): + def model_copy(self, *, update: dict[str, Any]) -> _Copy: + return _Copy(**{**vars(self), **update}) + + +def _lease(*, state: str = "HELD", expires_in: int = 300) -> _Copy: + return _Copy( + state=state, + expires_at=NOW + timedelta(seconds=expires_in), + process_id=1, + process_marker="m", + worker_id="w", + ) + + +class _Harness: + def __init__( + self, + *, + thread_state: ThreadExecutionState, + attempt_status: str, + lease: _Copy | None, + alive: bool | None = True, + ) -> None: + self.updates: list[Any] = [] + self.journaled: list[Any] = [] + self.replayed = 0 + thread = _Copy(execution_state=thread_state, revision=3) + attempt = _Copy(status=attempt_status, external_effect_state="NONE") + operation = SimpleNamespace(state=OperationState.RUNNING, project_id="p", operation_id="op") + + @contextmanager + def transaction() -> Any: + yield + + def update(value: Any, *, expected_revision: int) -> bool: + self.updates.append((value, expected_revision)) + return True + + def replay(_attempt: Any, _operation: Any) -> Any: + self.replayed += 1 + return SimpleNamespace(value={"replayed": True}, continue_execution=lambda: None) + + host = SimpleNamespace( + records=SimpleNamespace( + ledger=SimpleNamespace(transaction=transaction), + clock=SimpleNamespace(now=lambda: NOW), + journal_read=lambda *_a, **_k: lease, + journal=lambda *args: self.journaled.append(args), + ), + threads=SimpleNamespace(read=lambda _id: thread, update=update), + leases=SimpleNamespace(worker=SimpleNamespace(is_alive=lambda *_a: alive)), + accepted_value=lambda _attempt: {"operation_id": "op"}, + replay_attempt=replay, + ) + controls: Any = ResearchControls(host) # pyright: ignore[reportArgumentType] + controls.current = lambda _value: (attempt, operation) + self.controls = controls + + @property + def wrote(self) -> bool: + return bool(self.updates or self.journaled) + + +_PAUSED = ThreadExecutionState.PAUSED +_PENDING = ThreadExecutionState.PAUSE_PENDING +_IDLE = ThreadExecutionState.IDLE +_RUNNING = ThreadExecutionState.RUNNING + + +@pytest.mark.parametrize( + ("thread_state", "attempt_status", "lease", "unchanged"), + [ + (_PENDING, "PAUSE_REQUESTED", _lease(), True), # a pause is already pending + (_PAUSED, "PAUSED", None, True), # already paused, nothing running + (_IDLE, "RUNNING", _lease(), False), # running: the pause is a real change + (_PENDING, "RUNNING", _lease(), False), # thread says pending, attempt does not + ( + _PENDING, + "PAUSE_REQUESTED", + None, + False, + ), # pending but nothing runs: it settles to PAUSED + (_PAUSED, "PAUSED", _lease(), False), # paused thread but a worker holds the lease + ], +) +async def test_a_repeated_pause_is_unchanged_only_when_thread_and_attempt_already_match( + thread_state: ThreadExecutionState, + attempt_status: str, + lease: _Copy | None, + unchanged: bool, +) -> None: + harness = _Harness(thread_state=thread_state, attempt_status=attempt_status, lease=lease) + answered = await harness.controls.pause(SCOPE) + assert answered["pause_accepted"] is True + assert ("unchanged" in answered) is unchanged + assert harness.wrote is not unchanged + + +@pytest.mark.parametrize( + ("attempt_status", "lease", "alive", "unchanged"), + [ + ("RUNNING", _lease(), True, True), # running under a live worker + ("RUNNING", _lease(), False, False), # holder is dead: recover + ("RUNNING", _lease(), None, False), # liveness unknown: recover + ("RUNNING", _lease(expires_in=-1), True, False), # lease expired + ("RUNNING", _lease(state="RELEASED"), True, False), # lease released + ("RUNNING", None, True, False), # no lease at all + ("PAUSE_REQUESTED", _lease(), True, False), # a pause was requested: resume clears it + ("PAUSED", None, True, False), # paused: resume starts the run + ], +) +async def test_a_repeated_resume_is_unchanged_only_while_a_live_worker_runs_it( + attempt_status: str, lease: _Copy | None, alive: bool | None, unchanged: bool +) -> None: + harness = _Harness(thread_state=_IDLE, attempt_status=attempt_status, lease=lease, alive=alive) + answered = await harness.controls.resume(SCOPE) + if unchanged: + assert answered["unchanged"] is True + assert answered["unchanged_reason"] == "THREAD_ALREADY_RESUMED" + assert answered["resume_allowed"] is True and answered["execution_state"] == "IDLE" + assert not harness.wrote and harness.replayed == 0 + else: + # The recovery path: the thread is written and the attempt is dispatched again. + assert harness.updates and harness.journaled and harness.replayed == 1