From 8f44e146ba5ddb96de6f5d8a25ca36a423f5516a Mon Sep 17 00:00:00 2001 From: THOTH Date: Sun, 18 Oct 2026 23:43:00 +0900 Subject: [PATCH] Allow one THOTH process per local workspace; repeated state changes write nothing Two THOTH processes writing the same workspace could create duplicate record versions. Opening a local workspace's database now takes an exclusive lock on /runtime.lock before any migration or startup work. The operating system releases it when the process ends, however it ends, so a leftover file never blocks a restart. Runtimes inside one process share the lock. - serve and desktop take the lock before listening, so a second server is refused at start; a server started without the CLI answers 503 WORKSPACE_IN_USE on its first request. - shell, run, rpc, events and run-pack are refused with one Korean and one English line and exit code 3. Commands that do not open the database (workspace, doctor, source-stage, git-snapshot, auth-*) still run next to a running THOTH. - The process id and path in the refusal come from a hint in the lock file and are never stored. A refused start writes nothing to the first workspace. - Hosted review mode is not locked. Shared, synced and network folders are outside this protection (docs/KNOWN_LIMITATIONS.md). Repeating a state change that is already in effect now answers with the current state (unchanged=true and a reason) and writes no new record and no notification. Before, each repeat with a new request key wrote another revision and another notification. - Threads: pause, resume and stop. Investigations: pause and stop; a stop on an ended investigation no longer overwrites CONVERGED with STOPPED. - Closure decide and reopen, legacy baseline approval of the current baseline, and a repeated authorization vote while it is pending. - Research pause and resume. A resume is treated as a repeat only while the attempt is running under a lease whose worker is confirmed alive; a dead, unknown or expired worker still takes the recovery path. - A different decision, scope or request is still recorded as before. Tests pin the behavior before and after both changes, and how a request behaves when its process is killed mid-command (it stays RUNNING for the same key and nothing is written twice). Verification: the full local check (Python 5120 tests, web tests, lint, typecheck, build, architecture and module budget, README smoke) passed except one browser rendering test that fails only under heavy machine load. Its cause is pre-existing and unrelated to this change: closing the headless browser can exceed a 2 second limit and discard an already captured page. That fix follows separately. The test passes when rerun. --- README.ko.md | 2 + README.md | 2 + SOURCE_MANIFEST.json | 129 ++++-- config/module-responsibility-budget.json | 2 +- docs/INSTALL.md | 8 + docs/KNOWN_LIMITATIONS.ko.md | 14 +- docs/KNOWN_LIMITATIONS.md | 15 +- src/thoth/adapters/http/app.py | 15 +- src/thoth/adapters/storage/bundle.py | 40 +- src/thoth/adapters/storage/workspace_owner.py | 205 +++++++++ .../commands/action_mutation_projection.py | 10 +- .../application/commands/actions_full.py | 2 +- .../application/commands/closure_full.py | 35 +- .../application/commands/investigation.py | 24 +- .../commands/revision_publication_records.py | 17 + .../commands/revision_publication_views.py | 9 + .../application/commands/revisions_full.py | 53 +-- src/thoth/application/commands/threads.py | 29 ++ .../services/authorization_consumption.py | 8 + .../application/services/research_controls.py | 52 ++- src/thoth/apps/workspace_guard.py | 38 ++ src/thoth/cli.py | 124 +++--- src/thoth/domain/investigation.py | 4 + src/thoth/protocol/notifications.py | 5 + tests/integration/repeat_probe.py | 125 ++++++ .../test_stale_running_current_behavior.py | 342 +++++++++++++++ .../integration/test_state_transition_noop.py | 164 +++++++ ...tate_transition_repeat_current_behavior.py | 408 ++++++++++++++++++ ..._workspace_runtime_exclusivity_baseline.py | 224 ++++++++++ .../test_workspace_runtime_lock.py | 364 ++++++++++++++++ .../test_research_controls_repeat.py | 151 +++++++ 31 files changed, 2453 insertions(+), 167 deletions(-) create mode 100644 src/thoth/adapters/storage/workspace_owner.py create mode 100644 src/thoth/apps/workspace_guard.py create mode 100644 tests/integration/repeat_probe.py create mode 100644 tests/integration/test_stale_running_current_behavior.py create mode 100644 tests/integration/test_state_transition_noop.py create mode 100644 tests/integration/test_state_transition_repeat_current_behavior.py create mode 100644 tests/integration/test_workspace_runtime_exclusivity_baseline.py create mode 100644 tests/integration/test_workspace_runtime_lock.py create mode 100644 tests/unit/application/test_research_controls_repeat.py 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