diff --git a/.github/workflows/fork-safety.yml b/.github/workflows/fork-safety.yml new file mode 100644 index 0000000000..ec294ea4de --- /dev/null +++ b/.github/workflows/fork-safety.yml @@ -0,0 +1,26 @@ +name: Fork release safety + +on: + pull_request: + push: + branches: [develop] + workflow_dispatch: + +permissions: + contents: read + +jobs: + fork-release-safety: + name: Upstream updater protection + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v4 + - name: Verify fork updater is fail-closed + run: node scripts/fork/check-safety.mjs + - name: Verify fork desktop identity and local data roots + run: node scripts/fork/check-identity.mjs + - name: Verify JavaScript dev and numbered instance parity + run: node --test scripts/tauri/instance-profile.test.cjs + - name: Validate packaged two-tone icons + run: node scripts/fork/check-icons.mjs diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index e5c9ad4343..c10e1a5260 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -60,6 +60,8 @@ permissions: jobs: build-and-release: name: Build & Release (macOS aarch64) + # The upstream signing/release pipeline must never publish from a derivative fork. + if: ${{ github.repository == 'org2AI/ORG2' }} runs-on: macos-latest env: HAS_APPLE_CERT: ${{ secrets.APPLE_CERTIFICATE != '' }} @@ -248,6 +250,8 @@ jobs: # ── Windows x64 build ───────────────────────────────────────────────── build-windows: name: Build & Release (Windows x64) + # The upstream signing/release pipeline must never publish from a derivative fork. + if: ${{ github.repository == 'org2AI/ORG2' }} runs-on: windows-latest needs: build-and-release # This makes GitHub's OIDC subject stable across all release tags: @@ -469,6 +473,8 @@ jobs: # ── Linux x64 build ─────────────────────────────────────────────────── build-linux: name: Build & Release (Linux x64) + # The upstream signing/release pipeline must never publish from a derivative fork. + if: ${{ github.repository == 'org2AI/ORG2' }} runs-on: ubuntu-22.04 needs: build-windows steps: diff --git a/FORK.md b/FORK.md new file mode 100644 index 0000000000..250b8cf557 --- /dev/null +++ b/FORK.md @@ -0,0 +1,54 @@ +# Independent Fork Engineering Guide + +This repository is the independent development fork maintained at **TheerasakPing/ORG2**. +Upstream: https://github.com/org2AI/ORG2 (tracked on branch `develop`). +**TP Agent Studio** is the provisional product name. Fork-specific desktop IDs and paths are being introduced, but complete isolation and production release readiness are still unverified. + +## Current safety status + +- **Fork exists** with its own GitHub ownership and history. +- **Not independently distributable yet.** Do not install test builds side by side with upstream ORG2. +- The base Tauri updater has `plugins.updater.active=false`, an empty endpoints list and `bundle.createUpdaterArtifacts=false`. Fork README no longer links to upstream production installers. +- The custom Rust `check_app_update` command is also fail-closed while `FORK_UPDATER_READY=false`. This prevents silently installing upstream's release packages through a separate code path. +- Original two-tone desktop icon artwork and generated macOS/Windows/Linux icon assets are introduced for production and Dev, with the default macOS live Dock variant included (see [branding guide](docs/fork/TWO_TONE_BRANDING.md)). +- Base and development Tauri identities, deep-link schemes, macOS signing placeholders, application data defaults, runtime ports, and external history defaults have been moved to **TP Agent Studio**. **This does not yet prove complete profile isolation**: external tool hooks, sidecars, inherited assets/credentials and fresh-install behavior require integration tests. +- The inherited `release.yaml` workflow is guarded to run **only** in `org2AI/ORG2`, not this fork. Do not tag/publish production releases until a separate fork release workflow has been validated. + +## Fork strategy + +1. Maintain `develop` as the integration branch and use feature branches and PRs for every production change. +2. Keep attribution and the existing `LICENSE` intact. This project is under **AGPL-3.0-or-later**; derivative distribution and network use must comply with its source-availability and notice obligations. Do not imply that the original upstream developers endorse the fork. +3. Track upstream separately using `git remote add upstream https://github.com/org2AI/ORG2.git` in local clones. Import security and correctness fixes deliberately, preferably through reviewed merge commits; never overwrite fork-owned release, updater, storage, or identity settings. +4. Keep reusable code in focused modules, maintain regression tests, and avoid changing production invariants merely to hide UI defects. +5. Prioritize macOS Apple Silicon, Windows 11 x64, and Linux x64 parity in CI, packaging, and runtime acceptance tests. + +## Before enabling independent installation + +- Choose final product name, domains, app ID and deep-link schemes. Update main and platform-specific Tauri configurations together. +- Partition every mutable state root (SQLite, settings, account tokens, sidecars, caches, session data, logs, WebSocket local ports, single-instance control). Provide explicit opt-in migration from upstream with backup and rollback. +- Replace upstream update endpoints in both `src-tauri/tauri.conf.json` and `src-tauri/src/app_update.rs`, as well as any frontend release/download links; generate **fork-owned updater signing keys**. +- Replace upstream Apple signing identity, Windows signing configuration, GitHub environments, release URLs and permissions. Never reuse another publisher's private key. +- Add and pass fresh-install, side-by-side-install, upgrade, rollback, and uninstall-with-data-retention tests on each platform. +- Only after those gates pass: set `FORK_UPDATER_READY=true`, re-enable updater artifacts, and configure the fork's signed release channels. + +## Development and verification + +```bash +pnpm install --frozen-lockfile +pnpm run download:sidecars +pnpm run tauri:dev +node scripts/fork/check-safety.mjs +node scripts/fork/check-identity.mjs +pnpm typecheck +pnpm test +cd src-tauri +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace --no-fail-fast +``` + +CI file `.github/workflows/fork-safety.yml` checks the release-safety and desktop-identity posture independent of Rust/Node dependencies. The above commands are validation targets, **not claims that tests have passed on this fork**. + +## Contribution and upstream notice + +The original ORG2 source, licensing and attribution remain the foundation of this fork. +See [roadmap](docs/fork/ROADMAP.md). Any shipped modifications should record their author, purpose, test evidence and user-facing migration impact. diff --git a/README.md b/README.md index aa51f6ea90..1714e1b03c 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,33 @@ +> **Independent community fork — TheerasakPing/ORG2** +> +> This repository is a fork of [org2AI/ORG2](https://github.com/org2AI/ORG2). +> Work-in-progress app identity: **TP Agent Studio** (provisional branding; legacy compatibility names remain in internal APIs). +> +> ![TP Agent Studio two-tone icon](assets/branding/tp-agent-studio.svg) +> +> Two-tone desktop branding details: [branding guide](docs/fork/TWO_TONE_BRANDING.md). +> This fork does **not** distribute the upstream publisher's installers. +> The fork's application updater is intentionally disabled pending independent signing and release infrastructure. +> **Do not install experimental fork builds alongside upstream ORG2** until the app identity, +> deep links and `~/.orgii` state/storage are isolated and migration-tested. +> See [FORK.md](FORK.md), [the fork roadmap](docs/fork/ROADMAP.md), +> [Cloud boundary audit](docs/fork/FORK_CLOUD_BOUNDARY.md) and [session-stall recovery](docs/fork/SESSION_STALL_RECOVERY.md). +> +> **Development status (17 October 2026):** changes are integrated on [PR #9](https://github.com/TheerasakPing/ORG2/pull/9) +> against `develop`, **not yet merged or released**. CI must pass on the final commit, +> including Rust on Windows/macOS, before the pull request is merged. Platform installer, +> co-installation, memory/restart stress and fork-owned Cloud/OAuth acceptance tests remain outstanding. + +--- +
-

ORG-2

+

TP Agent Studio

The system of record for how agents build software.
Run your coding agents — replay any session, review as a team, and trace every line back to the decision behind it.

--- -

- macOS Apple Silicon - · - Windows installer - · - Windows MSI - · - Linux AppImage - · - Linux DEB - · - All latest release assets -

- +> **No TP Agent Studio installers are published yet.** macOS Apple Silicon, Windows 11, and Linux builds are undergoing identity, data-isolation, CI and signing verification. See [release readiness](FORK.md) and [engineering roadmap](docs/fork/ROADMAP.md). ---

@@ -37,6 +46,36 @@ ORG-2 is where your team runs its coding agents — a native Rust harness plus l It is not just another AI coding tool; it is an experiment in human/agent organizations and org-level alignment. ORG2 treats agents as persistent, observable colleagues inside a structured organization — replayable execution, cross-session memory, AI blame, and a local-first Rust runtime so humans, agents, and teams can collaborate around shared context and aligned goals. +## TP Agent Studio fork status / สถานะการพัฒนา + +**ภาษาไทย:** TP Agent Studio เป็น Fork อิสระของ ORG2 ซึ่งกำลังพัฒนาระบบ Rust Agent Harness, +การจัดการหลาย Session, Memory และการกู้คืนเมื่อการทำงานหยุดชะงัก ยังไม่มีไฟล์ติดตั้ง +ที่ผ่านการตรวจสอบสำหรับ macOS Apple Silicon (M4), Windows 11 หรือ Linux +และ **ยังไม่ควรใช้งานกับบัญชี Cloud จริงหรือข้อมูลที่สำคัญ** จนกว่าการแยกบริการ +OAuth/Cloud และทดสอบการย้ายข้อมูลจะเสร็จสมบูรณ์ + +### Recent work (integration PR #9) + +| Area | Implemented in the PR branch | Remaining verification | +| --- | --- | --- | +| Desktop identity | Independent TP Agent Studio package IDs, two-tone icons, fork deep-link scheme, dedicated profile roots and ports | Cross-platform install + side-by-side tests, remaining legacy resources | +| Rust Agent Harness / Session recovery | Per-session worker replacement after unexpected termination, future-factory panic isolation and persisted failure status | Heavy 1/5/20/50-session stress tests, long-hanging providers/tools, native E2E | +| Restart recovery | Continue independent startup DB reconciliation phases after partial errors; avoid automatically replaying side-effectful tools | Crash/kill/restart simulation on each OS and database integrity checks | +| Memory | Initial file-lease and fork-profile isolation hardening | Cross-process kill/restart proof, workspace-local migration, OS-specific validation | +| CI | Automated frontend, security, Rust and quality checks on every PR push | All checks green on **final** HEAD, including Windows Tauri icon parsing | +| Cloud / account isolation | Ownership risks documented; app updater and inherited upstream release workflow disabled | Fork-owned backend, callback allowlists, secure credentials, billing and release signing | + +This is **source-level hardening**, not a confirmed resolution for every multi-session freeze. +A never-resolving provider/tool call can still make a session look busy; do not delete +`sessions.db` or its WAL files to "unfreeze" the app. Use the non-destructive +[session-stall diagnostic and recovery guide](docs/fork/SESSION_STALL_RECOVERY.md). + +**Important:** GitHub commits on this fork do **not** modify an installed upstream ORG2 application. +To track code review and CI: [integration PR #9](https://github.com/TheerasakPing/ORG2/pull/9) · +[GitHub Actions](https://github.com/TheerasakPing/ORG2/actions) · +[Cloud/OAuth release blockers](docs/fork/FORK_CLOUD_BOUNDARY.md) · +[two-tone icon and regeneration](docs/fork/TWO_TONE_BRANDING.md). + ## Features @@ -190,20 +229,13 @@ Use ORG2's built-in Rust harness or launch these supported coding-agent CLIs fro Pi logo Pi

-## Download - -Current build version: v2.0.8 (2026-09-23) +## Release availability -Download the latest ORG2 desktop app with one click: +Current source version: v2.0.8 (2026-09-23) -- [macOS Apple Silicon](https://github.com/org2AI/ORG2/releases/latest/download/ORG2-latest-mac-apple-silicon.dmg) -- [Windows x64 installer](https://github.com/org2AI/ORG2/releases/latest/download/ORG2-latest-windows-x64-setup.exe) -- [Windows x64 MSI](https://github.com/org2AI/ORG2/releases/latest/download/ORG2-latest-windows-x64.msi) -- [Linux x64 AppImage](https://github.com/org2AI/ORG2/releases/latest/download/ORG2-latest-linux-x64.AppImage) -- [Linux x64 DEB](https://github.com/org2AI/ORG2/releases/latest/download/ORG2-latest-linux-x64.deb) -- [All latest release assets](https://github.com/org2AI/ORG2/releases/latest) +**No independently signed TP Agent Studio installers are available yet.** This fork is being developed and tested separately from the original publisher. Releases are blocked on green CI, Windows icon packaging, concurrency/restart stress testing, independent Cloud/OAuth and data/installer coexistence, and release signing. Builds will only be linked after macOS, Windows and Linux identity, migration, packaging and signature verification are complete. -The direct download links always resolve through GitHub's latest release pointer. +See [Fork Engineering Guide](FORK.md) for current safety status. Do not mistake upstream installers for releases of this fork. ## Develop from source diff --git a/assets/branding/tp-agent-studio-dev.svg b/assets/branding/tp-agent-studio-dev.svg new file mode 100644 index 0000000000..4269cc709a --- /dev/null +++ b/assets/branding/tp-agent-studio-dev.svg @@ -0,0 +1,13 @@ + + + + + + + + + + + + + diff --git a/assets/branding/tp-agent-studio.svg b/assets/branding/tp-agent-studio.svg new file mode 100644 index 0000000000..5a7e773479 --- /dev/null +++ b/assets/branding/tp-agent-studio.svg @@ -0,0 +1,10 @@ + + + + + + + + + + diff --git a/docs/fork/AUTO_DREAM_LOCK.md b/docs/fork/AUTO_DREAM_LOCK.md new file mode 100644 index 0000000000..9ed96bb1d8 --- /dev/null +++ b/docs/fork/AUTO_DREAM_LOCK.md @@ -0,0 +1,31 @@ +# Auto-Dream workspace memory — atomic lease upgrade + +## Why + +The original `.consolidate-lock` protocol wrote the process PID and then read it back. Two parallel tasks in the same process share a PID and could both believe they acquired the lock. On Windows the original PID liveness check always returned false, and mtime restoration was Unix-only. + +## Changes (stacked after Branding PR #3) + +- Add a distinct **permanent lock inode**: `{workspace}/.orgii/workspace-memory/.consolidate-lease`. +- Hold `fs2::FileExt::try_lock_exclusive` from before the legacy timestamp update until the RAII lease drops. This supplies OS-level advisory file locking across threads and processes on supported platforms. +- The old `.consolidate-lock` retains its previous timestamp+PID format so existing time-gates and persisted records continue to work. +- Replace Unix-specific timestamp rollback with cross-platform `std::fs::FileTimes`. +- Add eight-contender concurrency, same-process reentrancy, and subprocess lock exclusion tests alongside existing commit/rollback tests. + +## Invariants / remaining work + +1. Never remove or replace the `.consolidate-lease` file while another process may have it open. Keep its inode stable across app restarts; lock is released by the OS after process exit. +2. Only the RAII production entry point holds the lock. The timestamp/PID helper is now private; no product code should bypass the lease. +3. This change does **not** separate per-workspace `.orgii` memory from an upstream ORG2 instance opening the same project. Cross-product workspace metadata requires an opt-in compatibility/migration plan. +4. Crash consistency of the timestamp-based time-gate remains separate from exclusive ownership: a hard process crash may retain a too-recent timestamp; this will need a committed timestamp / in-progress marker split. +5. Run the included separate-process exclusion test on macOS Apple Silicon, Windows 11 and Linux; add *hard-kill/restart* tests and record successful results before shipping. + +## Focused check + +```bash +cd src-tauri +cargo test -p agent_core specialization::memory::workspace_memory::lock::tests +cargo clippy -p agent_core --all-targets -- -D warnings +``` + +**These are requested validation commands, not proof that CI has passed.** diff --git a/docs/fork/FORK_CLOUD_BOUNDARY.md b/docs/fork/FORK_CLOUD_BOUNDARY.md new file mode 100644 index 0000000000..1c158eaf41 --- /dev/null +++ b/docs/fork/FORK_CLOUD_BOUNDARY.md @@ -0,0 +1,54 @@ +# TP Agent Studio — Cloud and protocol ownership boundary + +**Status:** P0 release blocker, source audit only (2026-10-17). Not production-certified. +**PR:** https://github.com/TheerasakPing/ORG2/pull/9 + +## Verified from source + +| Area | Current fork behavior | Release implication | +| --- | --- | --- | +| Desktop protocol | Tauri production: `tpagentstudio://`; Dev: `tpagentstudio-dev://`; numbered: `tpagentstudio-instanceN://` | Native registration and per-OS fresh-install collision tests outstanding | +| Frontend runtime profile | `src/config/runtimeInstance.ts` now maps fork bundle IDs to the same 23847/27888 ports and fork protocols as Rust / JS launcher | Source parity is not proof of E2E callback registration | +| OAuth callback target | `configureCloudAuthCallbackForIdentifier` derives the target from the runtime bundle ID | External Cloud web server must explicitly allow its own fork redirect and validate state | +| Default cloud service | `src/features/Org2Cloud/config.ts` still embeds ORG2-managed Supabase and web origin as fallback | **No fork-owned Cloud has been configured; do not use real accounts or distribute as independent Cloud** | +| Endpoint override | LocalStorage key `orgii:org2-cloud-v1:endpointOverride` | Browser storage and keychain ownership must be audited; malformed override falls back to upstream | +| Inbound invite/share | `org2CloudOrgManagement.ts` accepts `orgii://cloud/...` | Fork URLs are not handled by dedicated parsers | +| Session reference | `cloudSessionReference.ts` uses the `orgii:` reference scheme | Cross-instance / fork-specific provenance undefined | +| Billing | `billingComplete.ts` accepts `orgii://billing/complete` | Must not reuse original publisher billing confirmation flow | +| Memory | Workspace-local `.orgii` remains compatible with upstream data | Avoid opening same workspace simultaneously until data isolation/migration is implemented | + +## Interim policy + +1. Keep updater, inherited release jobs, production signing, and publishing **disabled**. +2. Treat Cloud onboarding, billing, remote sync, auth callbacks and implicit upstream credentials as **not production-ready** on this fork. +3. Do not enable upstream Cloud defaults or loosen scheme parsers as a quick fix. Changing URL scheme strings alone does not create owned authentication, tenants or billing infrastructure. +4. Do not migrate or copy persisted upstream access/refresh tokens or OS keychain entries automatically. Require user consent, explicit source/target and rollback plan. +5. Continue to validate local-only Agent Harness, session UI and tools independently of Cloud. + +## Required implementation sequence + +### A — Inventory and fork-owned backend + +- [ ] Map all Cloud HTTP/WS clients, external login/payment pages, OS deep-link registrations, callback allowlists, app credentials and persisted sync cursors. +- [ ] Provision an explicitly fork-owned web origin and auth data plane, or an approved user-provided deployment, with isolated Supabase project, RLS, signing/key material and rotation process. +- [ ] Expose a release-safe Cloud **disabled/unconfigured** state instead of silently falling back to the upstream origin (including malformed or null overrides). Update UI so connection errors are actionable. +- [ ] Namespace Cloud browser keys/atoms, localStorage, OAuth state and secure OS credential storage per application + instance. +- [ ] Reject accidental imports from the upstream default unless a separately approved, scoped migration process authorizes them. + +### B — Owned protocol and parser contracts + +- [ ] Generate invite, share and billing callback links from an explicit, audited fork scheme and owned web origin. +- [ ] Normalize direct deep-link entry points for production, Dev and numbered instances; restrict auth and payment callbacks to exact expected scheme/host/path and state. +- [ ] Reject unexpected scheme, userinfo, duplicate critical parameters, fragments and malformed inputs. Never log or copy bearer tokens to generic navigation. +- [ ] Keep unsupported/unclaimed Cloud URLs fail-closed rather than converting to a generic React Router path. + +### C — Cross-platform acceptance + +- [ ] Unit-test explicit owned Cloud endpoint, missing config, malformed override, scheme matching, OAuth state, invite/share, billing, and data reset on switch. +- [ ] Test same-machine installs with original ORG2 on macOS Apple Silicon, Windows 11 and Linux: bundles, protocol handlers, ports, keychains, updater, workspace-local memory and credential files. +- [ ] Confirm genuine Cloud login and revocation against a **fork-owned staging tenant only** with redacted test artifacts. +- [ ] Record passing CI workflow URLs, installers/hashes, signed/notarized artifacts and a downgrade/rollback plan. + +**Acceptance:** the fork cannot automatically read, write, authenticate against, bill through or update itself from the original publisher's Cloud services or profiles. Cloud use remains explicitly disabled until this contract is met. + +Source findings are *not* operational proof; no Cloud tenancy, installer or credential-coexistence test is claimed here. diff --git a/docs/fork/MULTI_SESSION_STRESS_LAB.md b/docs/fork/MULTI_SESSION_STRESS_LAB.md new file mode 100644 index 0000000000..99c541cd3e --- /dev/null +++ b/docs/fork/MULTI_SESSION_STRESS_LAB.md @@ -0,0 +1,64 @@ +# TP Agent Studio — Multi-Session Stress Lab + +**Goal:** exercise the real Rust `DialogScheduler` with concurrent Session +workers under controlled, bounded loads. These are **synthetic, offline** +tests: no paid providers, actual tools, subprocesses or user credentials. + +## Run + +From the repository root: + +```bash +cd src-tauri +cargo test -p agent_core --lib session::scheduler::stress_tests -- --nocapture + +# One specific load profile, useful when debugging: +cargo test -p agent_core --lib session::scheduler::stress_tests::fifty_sessions_run_in_parallel_and_keep_fifo -- --exact --nocapture +``` + +A cold Rust compile can take significantly longer than the test execution. +Every scenario has a bounded 20-second in-runtime wait so a deadlocked +fixture fails rather than occupying CI indefinitely. Review the exit code. + +## Test matrix + +| Scenario | Synthetic load | Guarantee checked | +| --- | --- | --- | +| Baseline | 1 Session / 2 Turns | FIFO: second waits for first | +| Multi-5 | 5 Session workers / 10 Turns | Actual cross-session overlap; per-session FIFO | +| Multi-20 | 20 workers / 40 Turns | No cross-session serialization | +| Multi-50 | 50 workers / 100 Turns | All first Turns overlap, all second Turns drain once | +| Targeted cancellation | 5 active workers / 1 cancelled queued Turn | Cancellation stays within its Session | +| Worker abort | 2 workers, one force-aborted | Neighbor still finishes; lost work never replayed | +| Queue pressure | 1 busy worker, queue capacity 2 | Third pending message rejected; accepted work drains | + +The concurrency tests use actual Tokio task scheduling and +`DialogScheduler::enqueue()`. A barrier ensures that every first job +has reached a suspended execution boundary before any is released. +Per-session atomic state confirms the second job cannot run early. +The scheduler catches panics, so violations are recorded explicitly +rather than trusting panics to escape. + +## Important limitations + +- Passing 50 **synthetic** Tokio jobs does **not** prove 50 real provider + connections, high-token contexts or filesystem-changing tools are safe. +- No full-app frontend, sustained soak, process memory, real SQLite + contention, tool effects, crash/restart or installer acceptance is covered. +- A crashed worker's abandoned queued jobs are **not replayed automatically**. + Replaying side-effectful tool work could duplicate external actions. +- First-time Rust dependency compilation may dominate CI timing. Do not skip + failed tests or raise timeouts merely to obtain a green status. + +## Acceptance roadmap + +- [x] Add synthetic 1/5/20/50 scheduler profiles. +- [x] Add targeted-stop isolation, dead-worker isolation and queue overload. +- [ ] Verify focused Rust tests pass on CI (macOS/Windows/Linux). +- [ ] Fault-inject slow/failed providers, network, tools and SQLite contention. +- [ ] Validate 50 genuine Sessions with CPU, RAM, file descriptor and + latency telemetry over a 30-minute sustained workload. +- [ ] Verify full desktop shutdown/restart recovery and side-by-side + fork profile isolation on macOS M4, Windows 11 and Linux. + +See [Reliability roadmap](ROADMAP.md). diff --git a/docs/fork/PROFILE_ISOLATION.md b/docs/fork/PROFILE_ISOLATION.md new file mode 100644 index 0000000000..aef74cd05c --- /dev/null +++ b/docs/fork/PROFILE_ISOLATION.md @@ -0,0 +1,31 @@ +# Profile isolation: symlink-aware path guard + +## Background + +The independent **TP Agent Studio** binary uses its own default data home (`~/.tp-agent-studio`) and dedicated dev, numbered-instance and service-port namespaces. A portable/test environment can still override the internal `ORGII_HOME` setting. This override must **never** point the fork back into the original publisher's `~/.orgii` user profile. + +A previous guard directly called `canonicalize()` and, if it failed for a path that had not been created, compared the unnormalized string. A path with `../`, a symlinked ancestor or a non-existent suffix could evade the safety check. + +## Implemented + +- Require an absolute override path. +- Lexically resolve `.` and `..` path components before checking containment. +- Resolve the **nearest existing filesystem ancestor** and append missing components; if symlink resolution or permissions fail, refuse to start instead of silently accepting an unverified path. +- Compare to the independently resolved `~/.orgii` user data directory, including nested children and case-insensitive Windows path matching. +- Cover non-existent children, dot-dot traversal and symlinked parents in Rust tests; verify guard presence in the lightweight Fork Safety workflow. +- Maintain opt-in portable/test roots that do not overlap upstream's profile. + +## What is not yet guaranteed + +This is a startup **accidental-overlap guard**, not a hardened per-operation directory sandbox: an attacker who can change a symlink *after* startup can still cause a TOCTOU race. Workspace-local `.orgii` metadata, external CLI user profiles, OS credential stores and cloud accounts have separate paths and still need explicit auditing. Do not claim full fork/upstream isolation until those paths have been tested on macOS Apple Silicon, Windows 11 and Linux. + +## Verification + +```sh +node scripts/fork/check-identity.mjs +cd src-tauri +cargo test -p org2 runtime_instance::tests +cargo clippy --workspace --all-targets -- -D warnings +``` + +CI evidence, not this document, is the authority for a passing build. diff --git a/docs/fork/ROADMAP.md b/docs/fork/ROADMAP.md new file mode 100644 index 0000000000..0d46c99664 --- /dev/null +++ b/docs/fork/ROADMAP.md @@ -0,0 +1,75 @@ +# Independent ORG2 Fork — Engineering Roadmap + +**Status:** Planning and initial release-safety changes, not a production release. +**Repository:** https://github.com/TheerasakPing/ORG2 +**Upstream:** https://github.com/org2AI/ORG2 +**Tracking:** GitHub Issues are currently disabled on this fork; this file is the active task checklist until Issues are enabled. + +## P0 — Safe ownership and packaging (must pass before installers) + +- [ ] Select a final brand and domain; inventory ORG2 trademarks and upstream-only UI links. +- [x] Introduce original two-tone TP Agent Studio icons for desktop bundle, Dev and default live Dock; remaining tray, Store/mobile and legacy optional Dock variants need rebranding and platform validation. +- [x] Add provisional TP Agent Studio Tauri names/identifiers for main, dev and Windows; installer coexistence requires runtime acceptance tests. +- [x] Move the canonical application data root and runtime-default ports to fork-specific paths; external history and native transcript defaults are isolated. Remaining hard-coded legacy paths, runtime hooks and installer coexistence still require an end-to-end audit. +- [x] Disable base updater and custom Rust upstream update route for unreleased fork. +- [x] Remove clickable upstream installer links from fork README and clear the disabled Tauri updater manifest list. +- [ ] Replace remaining upstream public signing key, CI release dependencies and signing identities with fork-owned infrastructure. +- [x] Harden the explicit fork data-root guard against parent traversal, missing suffixes and symlinked ancestors; add focused Rust and fork-safety tests. +- [x] Align fork frontend runtime ports and OAuth callback schemes with embedded Rust application identifiers; include production/dev/numbered/E2E parity tests (CI evidence pending). +- [ ] **P0 Cloud ownership:** Fork defaults still reference the ORG2-managed Supabase/web backend and some Cloud parsers still accept only `orgii://` links. Configure fork-owned endpoints, secrets/credential namespaces and redirect allowlists; until then, prohibit real-account cloud onboarding and production distribution. See [Cloud boundary audit](FORK_CLOUD_BOUNDARY.md). +- [ ] Separate workspace-local `.orgii` metadata from upstream workspace metadata with opt-in migration, source-attributed ownership and rollback tests. +- [ ] Run same-machine concurrent-install and migration/rollback tests. +- [ ] Publish fork-built packages only after independent signing and end-to-end smoke tests. + +**Acceptance:** a fresh fork installation cannot update from upstream, overwrite upstream binaries, or read/write upstream profiles unless the user explicitly approves migration. + +## P1 — Reliability and security + +- [x] Replace the production auto-dream acquisition path with a stable OS-level exclusive file lease, plus eight-contender, cross-process exclusion tests and portable rollback. +- [ ] Add multi-process crash/kill/restart, Windows/macOS/Linux lock test evidence and split committed timestamp from in-progress state. +- [ ] Add sandbox profiles for untrusted agent shell commands; separate read/write paths, environment secrets, network and process limits. +- [ ] Introduce versioned, transactional SQLite migrations with verification and recovery procedures. +- [ ] Classify transient vs permanent LLM reflection failures; TTL/backoff rather than permanent blacklist for transient errors. +- [ ] Add session crash/restart, cancel/resume, provider-retry-after-tool and background-process teardown fault injection. +- [x] Catch scheduler execute-factory panics, track dead worker handles, recover the queue on next enqueue, and persist failed session state; added targeted regression tests (final CI pending). +- [x] Continue independent startup reconciliation phases after a partial failure so stale in-flight session cleanup is still attempted; see [multi-session recovery audit](SESSION_STALL_RECOVERY.md). +- [ ] Implement bounded progress/heartbeat diagnostics for genuinely hung provider or Tool futures, with cancellation semantics that never replay side-effectful work. +- [x] Add bounded, synthetic Rust `DialogScheduler` stress tests for 1/5/20/50 concurrent sessions, targeted cancellation, dead-worker no-replay isolation and queue capacity (test code added; CI verification pending). See [Multi-Session Stress Lab](MULTI_SESSION_STRESS_LAB.md). +- [ ] Stress-test 1/5/20/50 concurrent sessions under load and validate clean shutdown/restart on all three platforms. + +**Acceptance:** no double-owned memory consolidation; session/tool cancellation leaves no orphaned jobs; migration failures cannot leave an app silently running on a partially upgraded schema. + +## P2 — Unified agent experience + +- [ ] Define a shared session identity/event provenance contract for native Rust Harness and external CLI sessions. +- [ ] Preserve multi-agent task/call provenance across replay, forks, and resume. +- [ ] Build a unified runtime dashboard: tokens, costs, tool latency, retry reason, memory queue, session CPU/RAM, pending approvals. +- [ ] Add permissions preview and audit receipts for CLI, MCP, desktop and browser execution. +- [ ] Improve agent-team orchestration with role capabilities, checkpoints, budgets and human approval rules. + +**Acceptance:** imported and native sessions produce equivalent timeline events; each tool effect has an auditable identity; users can reproduce errors from structured diagnostics. + +## P3 — Product features and automation + +- [ ] Task templates, scheduling and unattended execution with max-cost and max-runtime limits. +- [ ] Agent workflows that plan, implement, test, review and prepare PRs under explicit branch protection. +- [ ] Memory provenance inspector with conflict handling, expiration and per-workspace privacy controls. +- [ ] Portable session export/import and cross-device encrypted synchronization (opt-in). +- [ ] Consistent UX and full localization; accessibility and performance budgets. + +## CI acceptance matrix + +| Platform | Build | Unit | Native integration | Install/upgrade | UI E2E | +| --- | --- | --- | --- | --- | --- | +| macOS Apple Silicon | Required | Required | Required | Required | Required | +| Windows 11 x64 | Required | Required | Required | Required | Required | +| Linux x64 | Required | Required | Required | Required | Required | + +Do not mark these checks complete based on source inspection; attach real logs, artifact hashes and CI run URLs. + +## Upstream sync policy + +- Prefer upstream security/fix updates before product-feature merges. +- Keep fork-specific release and identity configuration isolated in owned changes. +- For each upstream merge, rerun the fork-safety gate and release-install acceptance tests. +- Preserve `LICENSE`, copyright notices and upstream commit attribution. diff --git a/docs/fork/SESSION_STALL_RECOVERY.md b/docs/fork/SESSION_STALL_RECOVERY.md new file mode 100644 index 0000000000..70ba446e68 --- /dev/null +++ b/docs/fork/SESSION_STALL_RECOVERY.md @@ -0,0 +1,48 @@ +# Multi-session stall and crash-restart recovery + +**Status:** source-level reliability changes awaiting CI and high-load reproduction. This is **not** a certified fix for every stall. + +## Observed source-level failure paths + +1. `DialogScheduler::WorkerTask::run` originally invoked `(msg.execute)()` **outside** `catch_unwind`. A synchronous panic in that factory ended the entire per-session Tokio worker; later queued work could be stranded. +2. The scheduler retained a sender forever but did not check whether its worker had died, so a session could remain unusable without constructing a new session. +3. Executor errors could surface as `agent:error` while a SQLite session record retained `running`, producing a misleading sidebar state. +4. Startup recovery used `?` on the first phase failure, skipping the later, independent `running`-to-`abandoned` cleanup. A bad auxiliary turn-intent table could therefore leave phantom running sessions across restarts. +5. Long-lived futures that never resolve, OS resource exhaustion, third-party provider timeouts, and database contention are **not proven** to be caused by any one of the above. + +## Changes on integration PR #9 + +- Catch both future-factory and future-poll panics, returning a surfaced error without killing the session worker. +- Track the worker task; if it has ended unexpectedly, replace the receiver when the next item is enqueued. Explicitly discard orphaned in-memory messages rather than replaying potentially side-effectful tools. +- Reset queue counters, dedup and intent ownership on failed worker replacement; report the interrupted turn and persist `abandoned` best-effort. +- On ordinary executor failure without an already-terminal, stale, or user-intervention-owned intent, persist session `failed` to avoid phantom `running`. +- Attempt all startup recovery phases even after earlier errors, then aggregate/log failures. +- Add regression tests: synchronous factory panic followed by a succeeding turn; manually aborted idle worker replacement; failed executor clearing a durable `running` session. + +## Safe checks for an existing installed app + +**Do not delete `sessions.db`, its `-wal`/`-shm` companions, provider credentials, or entire app profiles to clear a stall.** The installed upstream ORG2 binary does not contain this fork patch. + +1. Quit the affected app normally. Preserve a copy of its profile folder, ideally after all related processes have exited, before any repair. For upstream this is generally `~/.orgii/`; for the fork's production build the intended default is `~/.tp-agent-studio/`. +2. Check app logs for `[scheduler]`, `[StartupRecovery]`, `database is locked`, `agent:error`, provider timeouts and out-of-memory messages. Redact tokens, URLs with credentials, user prompts and private paths before sharing. +3. If the `sqlite3` command is installed, inspect **read-only** status counts, e.g. after quitting: + ```sh + sqlite3 "file:$HOME/.orgii/sessions.db?mode=ro" \ + "SELECT status, COUNT(*) FROM agent_sessions GROUP BY status;" + ``` + Replace the path with the fork root if diagnosing a built TP Agent Studio. This does not edit or reset rows. +4. Capture exact app build version, OS, physical RAM, number of concurrently active sessions, provider/CLI type and whether creating a *fresh* session works. +5. Never force status changes with raw SQL on production user data. Recovery must use typed session lifecycle transitions and preserve terminal markers/Agent Org ownership. + +## Acceptance matrix (pending) + +- [ ] Reproduce simultaneous turns (1, 5, 20, 50) with real providers and separately with controlled local simulators. +- [ ] Verify a panic in the synchronous execute factory does not kill the worker or block later queued turns. +- [ ] Verify a killed/aborted worker is detectable, resets non-durable queue state, does not replay side-effectful tools, and accepts a fresh turn. +- [ ] Verify executor failure clears persisted `running` and emits one visible error event. +- [ ] Force auxiliary startup turn-intent recovery to fail and assert independent session status repair still runs. +- [ ] Test long provider timeouts, network disconnects, SQLite writer contention, high memory pressure, app close/kill/relaunch, and cancellation. +- [ ] Confirm macOS M4, Windows 11 and Linux CI, native E2E, side-by-side installation, app data isolation and successful rollback. +- [ ] Verify no automatic cloud login or upstream credential mixing (see `FORK_CLOUD_BOUNDARY.md`). + +**Release gate:** Do not merge as a production release or publish installer until the above behavioral and security evidence is recorded. Source modifications alone are not proof that the user's currently installed app is repaired. diff --git a/docs/fork/TWO_TONE_BRANDING.md b/docs/fork/TWO_TONE_BRANDING.md new file mode 100644 index 0000000000..43c7b176af --- /dev/null +++ b/docs/fork/TWO_TONE_BRANDING.md @@ -0,0 +1,30 @@ +# TP Agent Studio — Two-tone identity (provisional) + +![TP Agent Studio icon](../../assets/branding/tp-agent-studio.svg) + +## Palette and metaphor + +The fork's native desktop app uses **exactly two brand inks**: Midnight Navy `#101D33` and Electric Cyan `#4DD8F5` (alpha/transparency for shape edges is acceptable). The paired chevrons signify code/tools; the central node symbolizes agent orchestration. Do not reuse or imitate the upstream publisher's original icon or signing identity. + +## Source of truth and packaged resources + +- `assets/branding/tp-agent-studio.svg`: primary source vector, 512x512 viewBox. +- `assets/branding/tp-agent-studio-dev.svg`: same primary geometry with a small cyan development mark, distinguished by a navy center. +- `src-tauri/icons/{icon.png,icon.icns,icon.ico,32x32.png,64x64.png,128x128.png,128x128@2x.png}`: compiled assets for Windows, Linux and macOS. +- `src-tauri/icons/dev/{icon.png,icon.icns,icon.ico,32x32.png,128x128.png,128x128@2x.png}`: development icon variants. +- `src-tauri/icons/dock/dark.png`: uses the same primary art as the main application icon; this is the default live Dock icon. +- `scripts/fork/check-icons.mjs`: validates image signatures, dimensions, palette, PNG pixel compression, embedded ICO/ICNS images and the Dock invariant. + +## Regenerating + +SVG is the editable artwork. Rasterize to 8-bit RGBA PNG (not indexed PNG, which Tauri rejects at build time), package the PNG/ICO/ICNS resources, and run `node scripts/fork/check-icons.mjs`. The app's image validation is part of the fork-safety CI workflow. When changing the two color codes or path geometry, update both source SVGs and all bundled binary targets in the same pull request; do not ship the original ORG2 icon accidentally. + +## Remaining branding work + +- The historical `light` and `rainbow` optional Dock variants still use legacy upstream art; new fork-owned two-tone alternatives and settings migration are required before shipping. The default `dark` variant now uses TP Agent Studio art. +- Tray icon remains a platform-appropriate monochrome template; separately update its glyph before final release. +- Microsoft Store tile logos, mobile icons and UI splash/logo illustrations still need a dedicated rebranding pass. +- Update external URL/callback, cloud identity and public product copy only with ownership/compatibility tests. +- Perform installer, app-switcher and taskbar icon acceptance tests on macOS Apple Silicon, Windows 11 and Linux. + +**Safety:** fork updater remains disabled; no independently signed/distributed installers yet. Keep the upstream AGPL-3.0-or-later license and attribution. diff --git a/scripts/fork/check-icons.mjs b/scripts/fork/check-icons.mjs new file mode 100644 index 0000000000..637911dcdb --- /dev/null +++ b/scripts/fork/check-icons.mjs @@ -0,0 +1,119 @@ +#!/usr/bin/env node +// Validate the actual shipped two-tone desktop bitmaps, not only their source. +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import { resolve, dirname } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { inflateSync } from 'node:zlib'; + +const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); +const file = (p) => readFileSync(resolve(root, p)); +const sig = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]); +const navy = [16, 29, 51]; +const cyan = [77, 216, 245]; + +function assertRgbaPng(data, expectedSize, label) { + assert.ok(data.subarray(0, 8).equals(sig), label + ' missing PNG signature'); + let offset = 8, width, height, raw = []; + while (offset + 12 <= data.length) { + const n = data.readUInt32BE(offset), type = data.toString('ascii', offset + 4, offset + 8); + assert.ok(offset + n + 12 <= data.length, label + ' malformed PNG chunk'); + const payload = data.subarray(offset + 8, offset + 8 + n); + if (type === 'IHDR') { + width = payload.readUInt32BE(0); height = payload.readUInt32BE(4); + assert.equal(payload[8], 8, label + ' must use 8-bit channels'); + assert.equal(payload[9], 6, label + ' must use RGBA PNG, not indexed color'); + } + if (type === 'IDAT') raw.push(payload); + offset += 12 + n; + if (type === 'IEND') break; + } + if (expectedSize !== undefined) { + assert.equal(width, expectedSize, label + ' incorrect width'); + assert.equal(height, expectedSize, label + ' incorrect height'); + } + assert.ok(raw.length > 0, label + ' missing image data'); + const pixels = inflateSync(Buffer.concat(raw)); + const stride = width * 4 + 1; + assert.equal(pixels.length, stride * height, label + ' invalid pixel scanlines'); + let navyCount = 0, cyanCount = 0, transparentCount = 0; + for (let y = 0; y < height; y++) { + assert.equal(pixels[y * stride], 0, label + ' unexpected scanline filter'); + for (let x = 0; x < width; x++) { + const p = y * stride + 1 + x * 4; + const rgb = [...pixels.subarray(p, p + 3)], a = pixels[p + 3]; + if (a === 0) transparentCount++; + if (a === 255 && rgb.every((v, i) => v === navy[i])) navyCount++; + if (a === 255 && rgb.every((v, i) => v === cyan[i])) cyanCount++; + } + } + assert.ok(navyCount > 0 && cyanCount > 0 && transparentCount > 0, + label + ' must preserve both opaque brand inks and transparent corners'); + return data; +} + +function png(path, expectedSize) { + return assertRgbaPng(file(path), expectedSize, path); +} + +function ico(path) { + const data = file(path); + assert.equal(data.readUInt16LE(0), 0); + assert.equal(data.readUInt16LE(2), 1); + const count = data.readUInt16LE(4); + assert.ok(count >= 4, 'ICO lacks multi-resolution icons'); + for (let i = 0; i < count; i++) { + const entry = 6 + i * 16, len = data.readUInt32LE(entry + 8), offset = data.readUInt32LE(entry + 12); + assert.ok(offset + len <= data.length && len > 0, 'ICO image outside file'); + assert.ok(data.subarray(offset, offset + 8).equals(sig), 'ICO image not PNG'); + assertRgbaPng(data.subarray(offset, offset + len), undefined, path + ' frame ' + i); + } +} + +function icns(path) { + const data = file(path); + assert.equal(data.toString('ascii', 0, 4), 'icns'); + assert.equal(data.readUInt32BE(4), data.length); + let offset = 8, entries = 0; + while (offset < data.length) { + const len = data.readUInt32BE(offset + 4); + assert.ok(len > 8 && offset + len <= data.length, 'Malformed ICNS entry'); + assertRgbaPng(data.subarray(offset + 8, offset + len), undefined, path + ' entry ' + entries); + offset += len; entries++; + } + assert.ok(entries >= 6, 'ICNS lacks high-DPI variants'); +} + +const generated = [ + ['src-tauri/icons/32x32.png', 32], + ['src-tauri/icons/64x64.png', 64], + ['src-tauri/icons/128x128.png', 128], + ['src-tauri/icons/128x128@2x.png', 256], + ['src-tauri/icons/icon.png', 512], + ['src-tauri/icons/dev/32x32.png', 32], + ['src-tauri/icons/dev/128x128.png', 128], + ['src-tauri/icons/dev/128x128@2x.png', 256], + ['src-tauri/icons/dev/icon.png', 512], + ['src-tauri/icons/dock/dark.png', 512], +]; +for (const [path, n] of generated) png(path, n); +assert.ok(file('src-tauri/icons/icon.png').equals(file('src-tauri/icons/dock/dark.png')), + 'macOS live Dock must use the same mark as the installer'); +assert.ok(!file('src-tauri/icons/icon.png').equals(file('src-tauri/icons/dev/icon.png')), + 'Dev must be recognizable and distinguishable from production'); + +for (const path of ['src-tauri/icons/icon.ico', 'src-tauri/icons/dev/icon.ico']) ico(path); +for (const path of ['src-tauri/icons/icon.icns', 'src-tauri/icons/dev/icon.icns']) icns(path); + +for (const path of ['assets/branding/tp-agent-studio.svg', 'assets/branding/tp-agent-studio-dev.svg']) { + const source = file(path).toString('utf8'); + assert.ok(source.includes('#101D33') && source.includes('#4DD8F5'), + path + ' must preserve the two-ink brand palette'); +} +for (const configFile of ['src-tauri/tauri.conf.json', 'src-tauri/tauri.dev.conf.json']) { + const config = JSON.parse(file(configFile).toString('utf8')); + for (const icon of config.bundle.icon) { + file('src-tauri/' + icon); + } +} +console.log('Two-tone desktop icon sources, multi-resolution PNG, ICO and ICNS passed'); diff --git a/scripts/fork/check-identity.mjs b/scripts/fork/check-identity.mjs new file mode 100644 index 0000000000..525278f0ae --- /dev/null +++ b/scripts/fork/check-identity.mjs @@ -0,0 +1,63 @@ +#!/usr/bin/env node +// CI gate for fork desktop identity. No dependencies or network access. +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); +const read = (path) => readFileSync(resolve(root, path), 'utf8'); + +const base = JSON.parse(read('src-tauri/tauri.conf.json')); +const dev = JSON.parse(read('src-tauri/tauri.dev.conf.json')); +const windows = JSON.parse(read('src-tauri/tauri.windows.conf.json')); +const macos = JSON.parse(read('src-tauri/tauri.macos.conf.json')); + +assert.equal(base.identifier, 'com.theerasakping.tpagentstudio'); +assert.equal(dev.identifier, 'com.theerasakping.tpagentstudio.dev'); +assert.equal(base.productName, 'TP Agent Studio'); +assert.equal(dev.productName, 'TP Agent Studio Dev'); +assert.equal(base.app.windows.find((w) => w.label === 'main')?.title, base.productName); +assert.equal(windows.app.windows.find((w) => w.label === 'main')?.title, base.productName); +assert.deepEqual(base.plugins['deep-link'].desktop.schemes, ['tpagentstudio']); +assert.deepEqual(dev.plugins['deep-link'].desktop.schemes, ['tpagentstudio-dev']); +assert.equal(macos.bundle.macOS.signingIdentity, undefined, + 'Do not inherit upstream Apple signing identity'); +assert.equal(macos.bundle.macOS.providerShortName, undefined, + 'Do not inherit upstream Apple team short name'); +assert.equal(base.plugins.updater.active, false); +assert.equal(base.bundle.createUpdaterArtifacts, false); + +const scopes = base.app.security.assetProtocol.scope; +assert.ok(scopes.includes('$HOME/.tp-agent-studio/**')); +assert.ok(scopes.includes('$HOME/.tp-agent-studio-dev/**')); +assert.ok(!scopes.includes('$HOME/.orgii/**')); + +const nodeProfiles = read('scripts/tauri/instance-profile.cjs'); +assert.match(nodeProfiles, /const PRIMARY_IDE_PORT = 23847;/); +assert.match(nodeProfiles, /const PRIMARY_PROXY_PORT = 27888;/); +assert.match(nodeProfiles, /\.tp-agent-studio-dev/); +assert.match(nodeProfiles, /com\.theerasakping\.tpagentstudio/); + +const runtime = read('src-tauri/src/runtime_instance.rs'); +assert.match(runtime, /const FORK_IDE_SERVER_PORT: u16 = 23_847;/); +assert.match(runtime, /const FORK_CLI_PROXY_PORT: u16 = 27_888;/); +assert.match(runtime, /\.tp-agent-studio-dev/); +assert.match(runtime, /\.tp-agent-studio-instance/); +assert.match(runtime, /validate_explicit_home/); +assert.match(runtime, /resolve_profile_candidate/); +assert.match(runtime, /symlink_metadata/); +assert.match(runtime, /profile_path_contains/); +assert.match(runtime, /rejects_missing_suffix_and_parent_traversal_into_upstream_data/); +assert.match(runtime, /rejects_symlinked_parent_into_upstream_profile/); + +const bootstrap = read('src-tauri/src/app/bootstrap.rs'); +assert.match(bootstrap, /runtime_profile\.is_fork\(\)/); +assert.match(bootstrap, /validate_explicit_home/); +assert.match(bootstrap, /std::env::set_var\("ORGII_EXTERNAL_HISTORY_HOME", &history_home\)/); +assert.match(bootstrap, /std::env::set_var\("ORGII_NATIVE_TRANSCRIPT_HOME", &history_home\)/); +assert.match(read('src-tauri/crates/app-paths/src/data_root.rs'), + /home_dir\(\)\.join\("\.tp-agent-studio"\)/); +assert.match(read('src-tauri/src/app_update.rs'), + /const FORK_UPDATER_READY: bool = false;/); +process.stdout.write('Fork desktop identity checks passed. Full profile isolation still requires integration testing.\n'); diff --git a/scripts/fork/check-safety.mjs b/scripts/fork/check-safety.mjs new file mode 100644 index 0000000000..f7df1281e9 --- /dev/null +++ b/scripts/fork/check-safety.mjs @@ -0,0 +1,39 @@ +#!/usr/bin/env node +// Fork-specific guard. No dependencies or network access. +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import { dirname, resolve } from 'node:path'; + +const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); +const read = (path) => readFileSync(resolve(root, path), 'utf8'); + +const config = JSON.parse(read('src-tauri/tauri.conf.json')); +const rust = read('src-tauri/src/app_update.rs'); +const readme = read('README.md'); + +assert.equal(config.plugins?.updater?.active, false, + 'Fork must not activate Tauri updater until signed fork-only channels exist'); +assert.equal(config.bundle?.createUpdaterArtifacts, false, + 'Fork must not create unsigned or upstream-key updater artifacts'); +assert.match(rust, /const FORK_UPDATER_READY:\s*bool\s*=\s*false\s*;/, + 'Rust custom updater must stay fail-closed until independent publishing'); +assert.match(rust, /pub async fn check_app_update\([\s\S]*?require_fork_updater_ready\(\)\?;/, + 'Custom check_app_update path must enforce fork update guard'); +assert.match(readme, /Independent community fork/); +assert.deepEqual(config.plugins?.updater?.endpoints, [], + 'Until independent signing is configured, no updater manifest URL may be active'); +assert.ok(!readme.includes('https://github.com/org2AI/ORG2/releases/'), + 'Fork README must not offer binaries published by the original upstream'); +assert.match(readme, /No independently signed TP Agent Studio installers are available yet/); + +const upstreamRelease = read('.github/workflows/release.yaml'); +const releaseLines = upstreamRelease.split('\n'); +for (const job of ['build-and-release', 'build-windows', 'build-linux']) { + const anchor = releaseLines.findIndex((line) => line === ' ' + job + ':'); + assert.notEqual(anchor, -1, 'Missing release job: ' + job); + const block = releaseLines.slice(anchor, anchor + 8).join('\n'); + assert.ok(block.includes("if: ${{ github.repository == 'org2AI/ORG2' }}"), + 'Inherited upstream release job must be disabled in fork: ' + job); +} +process.stdout.write('Fork release-safety checks passed. Independent installation is not yet validated.\n'); diff --git a/scripts/tauri/instance-profile.cjs b/scripts/tauri/instance-profile.cjs index 940602556a..9d3be28ba3 100644 --- a/scripts/tauri/instance-profile.cjs +++ b/scripts/tauri/instance-profile.cjs @@ -1,8 +1,8 @@ const path = require("path"); const os = require("os"); -const PRIMARY_IDE_PORT = 13847; -const PRIMARY_PROXY_PORT = 17888; +const PRIMARY_IDE_PORT = 23847; +const PRIMARY_PROXY_PORT = 27888; const MAX_INSTANCE_ID = 99; // Reserve the next service slot for dev; numbered bundles remain 2..99. const DEV_INSTANCE_ID = 100; @@ -14,11 +14,11 @@ function createDevInstanceProfile() { productName: devConfig.productName, identifier: devConfig.identifier, deepLinkSchemes: devConfig.plugins["deep-link"].desktop.schemes, - authDeepLinkScheme: "orgii-dev", + authDeepLinkScheme: "tpagentstudio-dev", ideServerPort: PRIMARY_IDE_PORT + DEV_INSTANCE_ID - 1, cliProxyPort: PRIMARY_PROXY_PORT + DEV_INSTANCE_ID - 1, - // Share sessions.db and its related files with the bundled app. - dataHome: path.join(os.homedir(), ".orgii"), + // The fork dev build must never share sessions.db with production or upstream. + dataHome: path.join(os.homedir(), ".tp-agent-studio-dev"), }; } @@ -52,13 +52,13 @@ function createInstanceProfile(value) { const suffix = `instance${id}`; return { id, - productName: `ORG2 Instance ${id}`, - identifier: `org2ai.org2.${suffix}`, - deepLinkSchemes: [`yorgai-${suffix}`, `orgii-${suffix}`], - authDeepLinkScheme: `orgii-${suffix}`, + productName: `TP Agent Studio Instance ${id}`, + identifier: `com.theerasakping.tpagentstudio.${suffix}`, + deepLinkSchemes: [`tpagentstudio-${suffix}`], + authDeepLinkScheme: `tpagentstudio-${suffix}`, ideServerPort: PRIMARY_IDE_PORT + id - 1, cliProxyPort: PRIMARY_PROXY_PORT + id - 1, - dataHome: path.join(os.homedir(), `.orgii-${suffix}`), + dataHome: path.join(os.homedir(), `.tp-agent-studio-${suffix}`), }; } diff --git a/scripts/tauri/instance-profile.test.cjs b/scripts/tauri/instance-profile.test.cjs index 289399c0b8..aa08fdeedd 100644 --- a/scripts/tauri/instance-profile.test.cjs +++ b/scripts/tauri/instance-profile.test.cjs @@ -11,11 +11,11 @@ const { test("dev identity cannot collide with primary or any numbered bundle", () => { const dev = createDevInstanceProfile(); - assert.equal(dev.identifier, "org2ai.org2.dev"); - assert.equal(dev.ideServerPort, 13946); - assert.equal(dev.cliProxyPort, 17987); - assert.equal(dev.authDeepLinkScheme, "orgii-dev"); - assert.match(dev.dataHome, /\.orgii$/); + assert.equal(dev.identifier, "com.theerasakping.tpagentstudio.dev"); + assert.equal(dev.ideServerPort, 23946); + assert.equal(dev.cliProxyPort, 27987); + assert.equal(dev.authDeepLinkScheme, "tpagentstudio-dev"); + assert.match(dev.dataHome, /\.tp-agent-studio-dev$/); for (let id = 2; id <= 99; id++) { const bundle = createInstanceProfile(id); for (const key of [ @@ -38,9 +38,9 @@ test("dev frontend defaults agree with runtime ports and preserve explicit overr const env = { PATH: "/test/bin" }; assert.deepEqual(applyDevInstanceEnv(env), { PATH: "/test/bin", - ORGII_IDE_SERVER_PORT: "13946", - ORGII_CLI_PROXY_PORT: "17987", - ORGII_DEEP_LINK_SCHEME: "orgii-dev", + ORGII_IDE_SERVER_PORT: "23946", + ORGII_CLI_PROXY_PORT: "27987", + ORGII_DEEP_LINK_SCHEME: "tpagentstudio-dev", }); assert.deepEqual(env, { PATH: "/test/bin" }); const overrides = { @@ -55,11 +55,11 @@ test("dev frontend defaults agree with runtime ports and preserve explicit overr test("instance 2 receives isolated identity, schemes, and ports", () => { const profile = createInstanceProfile("2"); - assert.equal(profile.identifier, "org2ai.org2.instance2"); - assert.equal(profile.authDeepLinkScheme, "orgii-instance2"); - assert.equal(profile.ideServerPort, 13848); - assert.equal(profile.cliProxyPort, 17889); - assert.match(profile.dataHome, /\.orgii-instance2$/); + assert.equal(profile.identifier, "com.theerasakping.tpagentstudio.instance2"); + assert.equal(profile.authDeepLinkScheme, "tpagentstudio-instance2"); + assert.equal(profile.ideServerPort, 23848); + assert.equal(profile.cliProxyPort, 27889); + assert.match(profile.dataHome, /\.tp-agent-studio-instance2$/); }); test("primary and unbounded instance ids are rejected", () => { @@ -69,12 +69,12 @@ test("primary and unbounded instance ids are rejected", () => { }); test("IDE server ports resolve through the canonical instance profile", () => { - const profile = createInstanceProfileFromIdeServerPort("13848"); + const profile = createInstanceProfileFromIdeServerPort("23848"); assert.equal(profile.id, 2); - assert.equal(profile.productName, "ORG2 Instance 2"); - assert.equal(profile.cliProxyPort, 17889); + assert.equal(profile.productName, "TP Agent Studio Instance 2"); + assert.equal(profile.cliProxyPort, 27889); - for (const value of [undefined, "13847", "13946", "not-a-port"]) { + for (const value of [undefined, "23847", "23946", "not-a-port"]) { assert.throws(() => createInstanceProfileFromIdeServerPort(value)); } }); diff --git a/src-tauri/crates/agent-core/Cargo.toml b/src-tauri/crates/agent-core/Cargo.toml index 18c8abc4b9..84bd4d6771 100644 --- a/src-tauri/crates/agent-core/Cargo.toml +++ b/src-tauri/crates/agent-core/Cargo.toml @@ -78,6 +78,7 @@ schemars = { workspace = true } # --------------------------------------------------------------------------- rusqlite = { workspace = true } parking_lot = "0.12" +fs2 = "0.4" # Cross-process advisory lease for workspace memory consolidation. # --------------------------------------------------------------------------- # Time / IDs diff --git a/src-tauri/crates/agent-core/src/core/providers/traits.rs b/src-tauri/crates/agent-core/src/core/providers/traits.rs index 8b4bd8d9ee..5f524fdaa2 100644 --- a/src-tauri/crates/agent-core/src/core/providers/traits.rs +++ b/src-tauri/crates/agent-core/src/core/providers/traits.rs @@ -437,6 +437,8 @@ pub struct ChatOptions { /// /// Implementations handle the specifics of each provider's API /// while maintaining a consistent interface for the agent loop. +// async_trait expands async methods into boxed Futures already marked must_use. +#[allow(clippy::double_must_use)] #[async_trait] #[allow(clippy::too_many_arguments)] pub trait LLMProvider: Send + Sync { diff --git a/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs b/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs index a6ac27015e..b1bae25259 100644 --- a/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs +++ b/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs @@ -47,7 +47,7 @@ async fn native_ipc_embeds_disk_images_without_cli_refs_and_preserves_display_ro let sid = "sdeagent-transfer-images"; seed(sid); let url = "data:image/png;base64,QUJD".to_string(); - let id = super::super::save_user_msg(sid, "inspect", Some(&[url.clone()])).unwrap(); + let id = super::super::save_user_msg(sid, "inspect", Some(std::slice::from_ref(&url))).unwrap(); bind(sid, &id, "image-intent"); let stored = super::super::load_messages(sid).unwrap(); let paths: Vec = serde_json::from_str(stored[0].images.as_ref().unwrap()).unwrap(); @@ -60,7 +60,7 @@ async fn native_ipc_embeds_disk_images_without_cli_refs_and_preserves_display_ro assert_eq!(native[0]["turnIntentId"], "image-intent"); assert_eq!( native[0]["images"], - serde_json::to_string(&[url.clone()]).unwrap() + serde_json::to_string(std::slice::from_ref(&url)).unwrap() ); } let display = crate::state::commands::agent_load_messages(sid.into(), None) diff --git a/src-tauri/crates/agent-core/src/core/session/scheduler.rs b/src-tauri/crates/agent-core/src/core/session/scheduler.rs index f105275735..7b0fbee370 100644 --- a/src-tauri/crates/agent-core/src/core/session/scheduler.rs +++ b/src-tauri/crates/agent-core/src/core/session/scheduler.rs @@ -121,6 +121,45 @@ pub struct ScheduledMessage { pub execute: ExecuteFn, } +/// Decrement a pending-message counter only when nonzero, retaining +/// exactly the old checked_sub + fetch_update semantics on Rust MSRV 1.85. +fn decrement_pending(counter: &AtomicUsize) { + let mut current = counter.load(Ordering::Acquire); + while current > 0 { + match counter.compare_exchange_weak( + current, + current - 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return, + Err(actual) => current = actual, + } + } +} + +#[cfg(test)] +mod atomic_pending_tests { + use super::decrement_pending; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + #[test] + fn pending_counter_never_underflows() { + let count = Arc::new(AtomicUsize::new(7)); + let tasks: Vec<_> = (0..20) + .map(|_| { + let count = Arc::clone(&count); + std::thread::spawn(move || decrement_pending(&count)) + }) + .collect(); + for handle in tasks { + handle.join().unwrap(); + } + assert_eq!(count.load(Ordering::Acquire), 0); + } +} + pub(crate) fn panic_payload_to_string(payload: &(dyn Any + Send)) -> String { if let Some(message) = payload.downcast_ref::() { return message.clone(); @@ -181,6 +220,8 @@ pub struct EnqueueResult { /// Inner state initialized lazily on first enqueue. struct SchedulerInner { tx: mpsc::Sender, + /// Retain the worker handle to detect unexpected task termination. + worker: tokio::task::JoinHandle<()>, } /// Per-session FIFO message queue with a single background worker. @@ -253,10 +294,57 @@ impl DialogScheduler { async fn ensure_initialized(&self) -> mpsc::Sender { let mut guard = self.inner.lock().await; if let Some(inner) = guard.as_ref() { - return inner.tx.clone(); + if !inner.worker.is_finished() && !inner.tx.is_closed() { + return inner.tx.clone(); + } + + // A previous worker died (for example, from a panic outside the + // turn executor). Its receiver is gone, but the retained sender + // makes the scheduler appear initialized forever. Reset only the + // now-orphaned in-memory queue, never replay dropped tool effects. + warn!( + session_id = %self.session_id, + "[scheduler] worker terminated unexpectedly; replacing closed queue" + ); + self.generation.fetch_add(1, Ordering::AcqRel); + self.pending.store(0, Ordering::Release); + let was_running_turn = self.processing_turn.swap(false, Ordering::AcqRel); + self.processing.store(false, Ordering::Release); + self.client_message_ids.lock().await.clear(); + let interrupted = { + let mut control = self.turn_control.lock(); + control.cancelled.clear(); + control.current.take() + }; + if let Some(ref turn_id) = interrupted { + crate::foundation::session_bridge::update_turn_intent_status( + &self.session_id, + turn_id, + crate::foundation::session_bridge::TurnIntentBridgeStatus::Failed, + ); + } + crate::foundation::session_bridge::mark_pending_turn_intents_stale(&self.session_id); + if was_running_turn { + if let Err(err) = crate::session::persistence::update_status( + &self.session_id, + crate::session::SessionStatus::Abandoned, + ) { + warn!( + session_id = %self.session_id, + error = %err, + "[scheduler] could not persist interrupted session status" + ); + } + let streaming_error = StreamingError::new( + "Session worker stopped unexpectedly; the interrupted turn was not replayed. Send a new message to retry.".to_string(), + classify_streaming_error_message("Session worker stopped unexpectedly"), + ); + broadcast_agent_error_structured(&self.session_id, &streaming_error); + } + self.broadcast_queue_status(); } - // First enqueue — spawn the worker now + // First enqueue or replace a dead worker — spawn a new receiver let (tx, rx) = mpsc::channel::(self.capacity); let worker = WorkerTask { @@ -269,14 +357,17 @@ impl DialogScheduler { client_message_ids: Arc::clone(&self.client_message_ids), turn_control: Arc::clone(&self.turn_control), }; - tokio::spawn(worker.run()); + let worker = tokio::spawn(worker.run()); info!( "[scheduler] Worker spawned for session={} (capacity={})", self.session_id, self.capacity ); - *guard = Some(SchedulerInner { tx: tx.clone() }); + *guard = Some(SchedulerInner { + tx: tx.clone(), + worker, + }); tx } @@ -478,11 +569,7 @@ impl WorkerTask { } }; if targeted_cancelled { - let _ = self - .pending - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { - count.checked_sub(1) - }); + decrement_pending(&self.pending); crate::foundation::session_bridge::update_turn_intent_status( &self.session_id, &msg.turn_intent_id, @@ -524,11 +611,7 @@ impl WorkerTask { continue; } - let _ = self - .pending - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { - count.checked_sub(1) - }); + decrement_pending(&self.pending); let is_turn = msg.kind == ScheduledKind::Turn; self.processing.store(true, Ordering::Relaxed); @@ -566,8 +649,11 @@ impl WorkerTask { let client_message_id = msg.client_message_id.clone(); let turn_intent_id = msg.turn_intent_id.clone(); let org_run_id = msg.org_run_id.clone(); - let execute_future = (msg.execute)(); - let result = std::panic::AssertUnwindSafe(execute_future) + // The factory itself can panic before it returns a Future. + // Place its invocation INSIDE the catch boundary so an otherwise + // valid per-session worker cannot disappear with a live sender. + let execute = msg.execute; + let result = std::panic::AssertUnwindSafe(async move { execute().await }) .catch_unwind() .await .unwrap_or_else(|panic_payload| { @@ -693,6 +779,28 @@ impl WorkerTask { crate::foundation::session_bridge::TurnIntentBridgeStatus::Failed, ); } + // If the callback failed before its normal finalizer + // ran, the session DB can still say "running" while the + // in-memory worker has already moved on. Clear that + // phantom status for ordinary failed turns. Never alter + // previously terminal, stale, or intervention-owned work. + if is_turn + && !already_terminal + && !stale_callback + && !user_directed_waiting + && !user_directed_cancelled + { + if let Err(error) = crate::session::persistence::update_status( + &self.session_id, + crate::session::SessionStatus::Failed, + ) { + warn!( + session_id = %self.session_id, + error = %error, + "[scheduler] failed to persist failed turn session status" + ); + } + } // Turn-only: an `agent:error` renders as a chat bubble. // Maintenance jobs report failures through their own // channel (e.g. the manual-compact command's reply). @@ -821,6 +929,12 @@ async fn reconcile_agent_org_run_after_terminal(run_id: &str) { } } +#[cfg(test)] +mod stress_tests; + +#[cfg(test)] +mod fault_injection_tests; + #[cfg(test)] mod final_summary_tests; @@ -829,6 +943,161 @@ mod tests { use super::*; use std::sync::atomic::{AtomicUsize, Ordering}; + #[tokio::test] + async fn executor_factory_panic_does_not_kill_the_session_worker() { + let scheduler = DialogScheduler::new("session-factory-panic", 8); + scheduler + .enqueue(ScheduledMessage { + kind: ScheduledKind::Turn, + message_id: "factory-panics".into(), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: None, + content: "panic".into(), + execute: Box::new(|| panic!("factory failed before creating a future")), + }) + .await + .expect("factory panic must not occur on the caller"); + + let completed = Arc::new(AtomicUsize::new(0)); + let completed_in_turn = Arc::clone(&completed); + scheduler + .enqueue(ScheduledMessage { + kind: ScheduledKind::Turn, + message_id: "after-factory-panic".into(), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: None, + content: "continue".into(), + execute: Box::new(move || { + Box::pin(async move { + completed_in_turn.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }), + }) + .await + .expect("enqueue after factory panic"); + + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while completed.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("factory panic must not poison the worker or later queue items"); + assert_eq!(scheduler.pending_count(), 0); + } + + #[tokio::test] + #[serial_test::serial] + async fn failed_executor_reconciles_persisted_running_session() { + use crate::session::persistence::{self, UnifiedSessionRecord}; + use core_types::key_source::KeySource; + + let _sandbox = test_helpers::test_env::sandbox(); + let session_id = "session-failed-executor-status"; + { + let conn = database::db::get_connection().expect("test DB"); + crate::persistence::test_schema::ensure_agent_sessions_schema(&conn); + persistence::init(&conn).expect("session tables"); + } + let record = UnifiedSessionRecord { + session_id: session_id.to_string(), + name: "status regression".to_string(), + status: crate::session::SessionStatus::Running.as_str().to_string(), + created_at: "2024-01-01T00:00:00Z".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + session_type: "sde".to_string(), + key_source: KeySource::OwnKey, + ..Default::default() + }; + persistence::upsert_session(&record).expect("seed running session"); + + let scheduler = DialogScheduler::new(session_id, 8); + scheduler + .enqueue(ScheduledMessage { + kind: ScheduledKind::Turn, + message_id: "failing-turn".into(), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: None, + content: "fail".into(), + execute: Box::new(|| Box::pin(async { Err("provider failed".to_string()) })), + }) + .await + .expect("enqueue"); + + tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + if persistence::get_session(session_id) + .expect("query session") + .is_some_and(|row| row.status == crate::session::SessionStatus::Failed.as_str()) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("failed callback must clear durable running status"); + } + + #[tokio::test] + async fn a_dead_worker_is_recreated_on_the_next_enqueue() { + let scheduler = DialogScheduler::new("session-aborted-worker", 8); + scheduler.ensure_initialized().await; + { + let guard = scheduler.inner.lock().await; + guard.as_ref().expect("worker started").worker.abort(); + } + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let finished = { + let guard = scheduler.inner.lock().await; + guard.as_ref().expect("worker exists").worker.is_finished() + }; + if finished { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("aborted worker eventually stops"); + + let completed = Arc::new(AtomicUsize::new(0)); + let completed_in_turn = Arc::clone(&completed); + scheduler + .enqueue(ScheduledMessage { + kind: ScheduledKind::Maintenance, + message_id: "after-worker-abort".into(), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: None, + content: String::new(), + execute: Box::new(move || { + Box::pin(async move { + completed_in_turn.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }), + }) + .await + .expect("dead worker must be replaced"); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while completed.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("replacement worker processes a fresh queue item"); + } + #[test] fn targeted_cancellation_registry_is_bounded_and_keeps_the_latest_fence() { let scheduler = DialogScheduler::new("session-targeted-cancel-bound", 8); diff --git a/src-tauri/crates/agent-core/src/core/session/scheduler/fault_injection_tests.rs b/src-tauri/crates/agent-core/src/core/session/scheduler/fault_injection_tests.rs new file mode 100644 index 0000000000..a8dac8954c --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/session/scheduler/fault_injection_tests.rs @@ -0,0 +1,253 @@ +//! Offline fault-injection at the actual DialogScheduler future boundary. + //! No provider credentials, external HTTP calls or real tools are invoked. + //! Each test has an explicit, bounded deadline and no automatic replay. + +use super::{DialogScheduler, ExecutionCompletion, ScheduledKind, ScheduledMessage}; +use std::future::pending; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::oneshot; + +const DEADLINE: Duration = Duration::from_secs(10); + +fn job( + label: &str, + execute: super::ExecuteFn, +) -> ScheduledMessage { + ScheduledMessage { + kind: ScheduledKind::Maintenance, + message_id: label.to_string(), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: None, + content: String::new(), + execute, + } +} + +/// A provider callback that is waiting on a remote reply must not cause +/// unrelated schedulers to block. A follow-up in the *same* FIFO must wait. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn slow_provider_future_isolated_from_five_neighbor_sessions() { + let blocked = DialogScheduler::new("slow-provider", 4); + let (started_tx, started_rx) = oneshot::channel::<()>(); + let (release_tx, release_rx) = oneshot::channel::<()>(); + let provider_executed = Arc::new(AtomicUsize::new(0)); + let provider_count = Arc::clone(&provider_executed); + blocked.enqueue(job("waiting-provider", Box::new(move || { + Box::pin(async move { + let _ = started_tx.send(()); + release_rx.await.map_err(|_| "fixture release dropped".to_string())?; + provider_count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("provider fixture enqueued"); + tokio::time::timeout(DEADLINE, started_rx) + .await.expect("provider callback must enter").expect("start signal"); + + let followup_finished = Arc::new(AtomicUsize::new(0)); + let followup_count = Arc::clone(&followup_finished); + blocked.enqueue(job("queued-behind-provider", Box::new(move || { + Box::pin(async move { + followup_count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("follow-up must enqueue"); + + let other_finished = Arc::new(AtomicUsize::new(0)); + let mut neighbors = Vec::new(); + for n in 0..5 { + let scheduler = DialogScheduler::new(format!("provider-neighbor-{n}"), 2); + let finished = Arc::clone(&other_finished); + scheduler.enqueue(job("unrelated-session", Box::new(move || { + Box::pin(async move { + finished.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("independent worker enqueues"); + neighbors.push(scheduler); + } + tokio::time::timeout(DEADLINE, async { + while other_finished.load(Ordering::SeqCst) != 5 { + tokio::task::yield_now().await; + } + }).await.expect("other workers must complete while provider waits"); + assert!(blocked.is_processing(), "slow future is still owned by its worker"); + assert_eq!(blocked.pending_count(), 1); + assert_eq!(provider_executed.load(Ordering::SeqCst), 0); + assert_eq!(followup_finished.load(Ordering::SeqCst), 0); + release_tx.send(()).expect("blocked callback remains pending"); + tokio::time::timeout(DEADLINE, async { + while followup_finished.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }).await.expect("follow-up must drain once original provider completes"); + assert_eq!(provider_executed.load(Ordering::SeqCst), 1); + assert_eq!(followup_finished.load(Ordering::SeqCst), 1); + assert_eq!(blocked.pending_count(), 0); + assert_eq!(neighbors.len(), 5); +} + +/// An explicit mock/provider timeout must return a bounded error. The +/// scheduler must not poison the FIFO after receiving that error. +/// Maintenance kind isolates the scheduling contract from durable session +/// state and does not claim to test production retry decisions. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn explicit_mock_provider_timeout_does_not_poison_followup_queue() { + let scheduler = DialogScheduler::new("timeout-provider", 4); + let timed_out = Arc::new(AtomicUsize::new(0)); + let errors = Arc::clone(&timed_out); + scheduler.enqueue(job("mock-provider-timeout", Box::new(move || { + Box::pin(async move { + let outcome = tokio::time::timeout( + Duration::from_millis(60), + pending::<()>(), + ).await; + if outcome.is_err() { + errors.fetch_add(1, Ordering::SeqCst); + return Err("synthetic-provider-timeout".to_string()); + } + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("provider test enqueued"); + + let continued = Arc::new(AtomicUsize::new(0)); + let finished = Arc::clone(&continued); + scheduler.enqueue(job("after-timeout", Box::new(move || { + Box::pin(async move { + finished.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("next job accepted"); + + tokio::time::timeout(DEADLINE, async { + while continued.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }).await.expect("worker must keep draining after explicit simulated error"); + assert_eq!(timed_out.load(Ordering::SeqCst), 1); + assert_eq!(continued.load(Ordering::SeqCst), 1); + assert_eq!(scheduler.pending_count(), 0); +} + +/// An indefinite tool future cannot be replayed by an observer. Aborting +/// its scheduler task drops the in-flight future, and the replacement must +/// accept only a *fresh* job. A neighboring scheduler remains independent. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn never_resolving_tool_is_dropped_on_worker_abort_without_replay() { + struct DropCount(Arc); + impl Drop for DropCount { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + + let blocked = DialogScheduler::new("hanging-tool", 4); + let healthy = DialogScheduler::new("healthy-neighbor", 4); + let drop_count = Arc::new(AtomicUsize::new(0)); + let (started_tx, started_rx) = oneshot::channel::<()>(); + let tool_drop = Arc::clone(&drop_count); + + blocked.enqueue(job("never-return-tool", Box::new(move || { + Box::pin(async move { + let _watch = DropCount(tool_drop); + let _ = started_tx.send(()); + pending::<()>().await; + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("pending tool job accepted"); + tokio::time::timeout(DEADLINE, started_rx) + .await.expect("pending tool must start") + .expect("start notification"); + + let lost_queued_executed = Arc::new(AtomicUsize::new(0)); + let lost = Arc::clone(&lost_queued_executed); + blocked.enqueue(job("lost-queued-followup", Box::new(move || { + Box::pin(async move { + lost.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("pending followup accepted"); + + let neighbor_executed = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&neighbor_executed); + healthy.enqueue(job("healthy-followup", Box::new(move || { + Box::pin(async move { + observed.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("neighbor work accepted"); + tokio::time::timeout(DEADLINE, async { + while neighbor_executed.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }).await.expect("neighbor must be able to finish"); + + { + let owner = blocked.inner.lock().await; + owner.as_ref().expect("worker exists").worker.abort(); + } + tokio::time::timeout(DEADLINE, async { + loop { + let finished = { + let owner = blocked.inner.lock().await; + owner.as_ref().expect("worker exists").worker.is_finished() + }; + if finished && drop_count.load(Ordering::SeqCst) == 1 { break; } + tokio::task::yield_now().await; + } + }).await.expect("aborted worker must drop its nonterminating tool"); + + let fresh_executed = Arc::new(AtomicUsize::new(0)); + let fresh = Arc::clone(&fresh_executed); + blocked.enqueue(job("fresh-after-abort", Box::new(move || { + Box::pin(async move { + fresh.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("fresh enqueue triggers recovery"); + tokio::time::timeout(DEADLINE, async { + while fresh_executed.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }).await.expect("fresh job must complete"); + assert_eq!(drop_count.load(Ordering::SeqCst), 1); + assert_eq!(lost_queued_executed.load(Ordering::SeqCst), 0, + "previously queued side effects must not be replayed"); + assert_eq!(fresh_executed.load(Ordering::SeqCst), 1); + assert_eq!(neighbor_executed.load(Ordering::SeqCst), 1); + assert_eq!(blocked.pending_count(), 0); +} + +/// Tool errors should be scoped to one callback. The worker must keep +/// draining subsequent work; error text is synthetic and content-free. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn mock_tool_error_does_not_starve_following_job() { + let scheduler = DialogScheduler::new("tool-error", 2); + let errors = Arc::new(AtomicUsize::new(0)); + let errors_in_job = Arc::clone(&errors); + scheduler.enqueue(job("failing-tool", Box::new(move || { + Box::pin(async move { + errors_in_job.fetch_add(1, Ordering::SeqCst); + Err("synthetic-tool-error".to_string()) + }) + }))).await.expect("failed tool fixture enqueued"); + + let completed = Arc::new(AtomicUsize::new(0)); + let after = Arc::clone(&completed); + scheduler.enqueue(job("after-failure", Box::new(move || { + Box::pin(async move { + after.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }))).await.expect("second job accepted"); + tokio::time::timeout(DEADLINE, async { + while completed.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }).await.expect("failed job must not stop queue"); + assert_eq!(errors.load(Ordering::SeqCst), 1); + assert_eq!(completed.load(Ordering::SeqCst), 1); +} diff --git a/src-tauri/crates/agent-core/src/core/session/scheduler/stress_tests.rs b/src-tauri/crates/agent-core/src/core/session/scheduler/stress_tests.rs new file mode 100644 index 0000000000..9e28445d69 --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/session/scheduler/stress_tests.rs @@ -0,0 +1,386 @@ +//! Synthetic, offline stress tests of the actual DialogScheduler. + //! + //! No real LLM/network/tool calls, account credentials, or interactive UI. + //! All tests are bounded: use 1/5/20/50 session profiles and finite deadlines. + //! These checks prove FIFO and cross-session isolation at the Scheduler + //! boundary, NOT end-to-end provider liveness, crash recovery, or production + //! capacity under 50 real concurrent API/tool workloads. + +use super::{DialogScheduler, ExecutionCompletion, ScheduledKind, ScheduledMessage}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::{watch, Barrier}; + +const TEST_DEADLINE: Duration = Duration::from_secs(20); + +fn synthetic_message( + message_id: String, + kind: ScheduledKind, + execute: super::ExecuteFn, +) -> ScheduledMessage { + ScheduledMessage { + kind, + message_id, + generation: 0, + client_message_id: None, + // Empty as in the existing scheduler unit tests. This is an in-memory + // concurrency test, not the persistent Turn-intent identity contract. + turn_intent_id: String::new(), + org_run_id: None, + content: String::new(), + execute, + } +} + +/// Concurrent native session workers must not serialize across sessions. +/// The first Turn in each FIFO reaches a barrier before any can finish. +/// Then, a second Turn per session must not start before its first one ends. +async fn exercise_profile(session_count: usize) { + let barrier = Arc::new(Barrier::new(session_count + 1)); + let (release, hold) = watch::channel(false); + let active = Arc::new(AtomicUsize::new(0)); + let first_started = Arc::new(AtomicUsize::new(0)); + let first_completed = Arc::new(AtomicUsize::new(0)); + let second_completed = Arc::new(AtomicUsize::new(0)); + let ordering_errors = Arc::new(AtomicUsize::new(0)); + let mut schedulers = Vec::with_capacity(session_count); + let mut states = Vec::with_capacity(session_count); + + for index in 0..session_count { + let scheduler = DialogScheduler::new( + format!("stress-lab-{session_count}-{index}"), + 4, // 1 queued follow-up + generous per-session queue headroom + ); + let state = Arc::new(AtomicUsize::new(0)); + states.push(Arc::clone(&state)); + let initial_state = Arc::clone(&state); + let first_barrier = Arc::clone(&barrier); + let mut hold_rx = hold.clone(); + let active_in_job = Arc::clone(&active); + let started_in_job = Arc::clone(&first_started); + let completed_in_job = Arc::clone(&first_completed); + let errors_in_job = Arc::clone(&ordering_errors); + + scheduler + .enqueue(synthetic_message( + format!("first-{index}"), + ScheduledKind::Turn, + Box::new(move || { + Box::pin(async move { + active_in_job.fetch_add(1, Ordering::SeqCst); + started_in_job.fetch_add(1, Ordering::SeqCst); + // Once all N first jobs have reached the barrier, + // cross-session scheduling has demonstrably overlapped. + first_barrier.wait().await; + loop { + // Drop the watch Ref before awaiting a mutable + // changed() call (avoids a borrow across await). + let released = { *hold_rx.borrow() }; + if released { break; } + if hold_rx.changed().await.is_err() { + return Err("stress-lab release channel was dropped".into()); + } + } + if initial_state.swap(1, Ordering::SeqCst) != 0 { + errors_in_job.fetch_add(1, Ordering::SeqCst); + } + active_in_job.fetch_sub(1, Ordering::SeqCst); + completed_in_job.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }), + )) + .await + .expect("first Turn must enqueue"); + + let second_state = Arc::clone(&state); + let second_done = Arc::clone(&second_completed); + let second_errors = Arc::clone(&ordering_errors); + scheduler + .enqueue(synthetic_message( + format!("second-{index}"), + ScheduledKind::Turn, + Box::new(move || { + Box::pin(async move { + // A violation is counted even if scheduler catches a + // panic and otherwise masks it as a Turn error. + if second_state.swap(2, Ordering::SeqCst) != 1 { + second_errors.fetch_add(1, Ordering::SeqCst); + } + second_done.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + }) + }), + )) + .await + .expect("second FIFO Turn must enqueue"); + // Keep the real session-owned scheduler alive until all work drains. + schedulers.push(scheduler); + } + + tokio::time::timeout(TEST_DEADLINE, barrier.wait()) + .await + .expect("all first Turns must reach the cross-session barrier"); + assert_eq!(first_started.load(Ordering::SeqCst), session_count); + assert_eq!(active.load(Ordering::SeqCst), session_count); + assert_eq!(second_completed.load(Ordering::SeqCst), 0, + "no FIFO follow-up may start while its first Turn is blocked"); + + release.send(true).expect("there must still be live Turn receivers"); + tokio::time::timeout(TEST_DEADLINE, async { + while second_completed.load(Ordering::SeqCst) != session_count { + tokio::task::yield_now().await; + } + }) + .await + .expect("all second FIFO Turns must eventually complete"); + assert_eq!(first_completed.load(Ordering::SeqCst), session_count); + assert_eq!(ordering_errors.load(Ordering::SeqCst), 0); + assert_eq!(active.load(Ordering::SeqCst), 0); + for (index, state) in states.iter().enumerate() { + assert_eq!(state.load(Ordering::SeqCst), 2, + "both FIFO turns must have executed exactly once for session {index}"); + } + for scheduler in &schedulers { + assert_eq!(scheduler.pending_count(), 0); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn one_session_preserves_fifo() { + exercise_profile(1).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn five_sessions_run_in_parallel_and_keep_fifo() { + exercise_profile(5).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn twenty_sessions_run_in_parallel_and_keep_fifo() { + exercise_profile(20).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fifty_sessions_run_in_parallel_and_keep_fifo() { + exercise_profile(50).await; +} + +/// A failure/targeted Stop in one session cannot cancel a neighbor's queued +/// Turn, even if all sessions are actively processing first Turns. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn targeted_stop_is_session_local_across_five_busy_workers() { + const COUNT: usize = 5; + const STOP_INDEX: usize = 2; + let barrier = Arc::new(Barrier::new(COUNT + 1)); + let (release, hold) = watch::channel(false); + let followup_count = Arc::new(AtomicUsize::new(0)); + let cancelled_ran = Arc::new(AtomicUsize::new(0)); + let mut schedulers = Vec::new(); + + for index in 0..COUNT { + let scheduler = DialogScheduler::new( + format!("stop-isolation-{index}"), + 4, + ); + let b = Arc::clone(&barrier); + let mut rx = hold.clone(); + scheduler.enqueue(synthetic_message( + format!("first-{index}"), + ScheduledKind::Turn, + Box::new(move || Box::pin(async move { + b.wait().await; + loop { + let released = { *rx.borrow() }; + if released { break; } + if rx.changed().await.is_err() { + return Err("release was dropped".into()); + } + } + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("first Turn enqueued"); + + // Unlike the empty intent of the first fixture, targeted cancellation + // must have a uniquely owned exact intent key. + let second_id = format!("cancel-target-{index}"); + let finished = Arc::clone(&followup_count); + let cancelled = Arc::clone(&cancelled_ran); + scheduler.enqueue(ScheduledMessage { + kind: ScheduledKind::Turn, + message_id: format!("followup-{index}"), + generation: 0, + client_message_id: None, + turn_intent_id: second_id.clone(), + org_run_id: None, + content: String::new(), + execute: Box::new(move || Box::pin(async move { + if index == STOP_INDEX { + cancelled.fetch_add(1, Ordering::SeqCst); + } else { + finished.fetch_add(1, Ordering::SeqCst); + } + Ok(ExecutionCompletion::Finished) + })), + }).await.expect("second Turn enqueued"); + if index == STOP_INDEX { + assert!(!scheduler.invalidate_turn(&second_id), + "second Turn is still pending while first is blocked"); + } + schedulers.push(scheduler); + } + tokio::time::timeout(TEST_DEADLINE, barrier.wait()) + .await.expect("all five initial Turns should start"); + release.send(true).expect("receivers alive"); + tokio::time::timeout(TEST_DEADLINE, async { + while followup_count.load(Ordering::SeqCst) != COUNT - 1 { + tokio::task::yield_now().await; + } + }).await.expect("unaffected sessions must finish"); + // The cancelled job can still be waiting for its scheduler's receiver + // while other sessions complete. Wait for its dequeue without sleeping. + tokio::time::timeout(TEST_DEADLINE, async { + while schedulers.iter().any(|worker| worker.pending_count() != 0) { + tokio::task::yield_now().await; + } + }).await.expect("targeted cancelled job must drain its queue slot"); + assert_eq!(cancelled_ran.load(Ordering::SeqCst), 0); +} + + +/// The worker for one session can be aborted without blocking an unrelated +/// per-session FIFO. Replacement only happens on a *new enqueue* and never +/// replays the old pending job from the lost receiver. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn dead_worker_replacement_does_not_replay_or_block_neighbor() { + let dead = DialogScheduler::new("stress-lab-dead", 4); + let neighbor = DialogScheduler::new("stress-lab-neighbor", 4); + let (dead_started_tx, dead_started_rx) = tokio::sync::oneshot::channel(); + let (_dead_release_tx, dead_release_rx) = tokio::sync::oneshot::channel::<()>(); + let stale_ran = Arc::new(AtomicUsize::new(0)); + let recovered_ran = Arc::new(AtomicUsize::new(0)); + let neighbor_ran = Arc::new(AtomicUsize::new(0)); + + dead.enqueue(synthetic_message( + "dead-running".into(), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + let _ = dead_started_tx.send(()); + let _ = dead_release_rx.await; + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("enqueue blocked worker"); + tokio::time::timeout(TEST_DEADLINE, dead_started_rx) + .await.expect("dead worker starts") + .expect("start signal"); + + let stale_count = Arc::clone(&stale_ran); + dead.enqueue(synthetic_message( + "never-replay".into(), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + stale_count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("enqueue old pending job"); + + let (neighbor_release, neighbor_hold) = tokio::sync::oneshot::channel::<()>(); + let (neighbor_started_tx, neighbor_started_rx) = tokio::sync::oneshot::channel(); + let neighbor_count = Arc::clone(&neighbor_ran); + neighbor.enqueue(synthetic_message( + "neighbor".into(), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + let _ = neighbor_started_tx.send(()); + let _ = neighbor_hold.await; + neighbor_count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("neighbor starts"); + tokio::time::timeout(TEST_DEADLINE, neighbor_started_rx) + .await.expect("neighbor is independently active") + .expect("neighbor start signal"); + + { + let guard = dead.inner.lock().await; + guard.as_ref().expect("dead worker registered").worker.abort(); + } + tokio::time::timeout(TEST_DEADLINE, async { + loop { + let finished = dead.inner.lock().await + .as_ref().expect("dead handle retained").worker.is_finished(); + if finished { break; } + tokio::task::yield_now().await; + } + }).await.expect("aborted worker must stop"); + + let recovered_count = Arc::clone(&recovered_ran); + dead.enqueue(synthetic_message( + "new-work-not-replay".into(), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + recovered_count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("a new enqueue must replace terminated worker"); + + neighbor_release.send(()).expect("neighbor is still waiting"); + tokio::time::timeout(TEST_DEADLINE, async { + while recovered_ran.load(Ordering::SeqCst) != 1 || + neighbor_ran.load(Ordering::SeqCst) != 1 { + tokio::task::yield_now().await; + } + }).await.expect("new local job and neighbor job both finish"); + + assert_eq!(stale_ran.load(Ordering::SeqCst), 0, + "lost queue contents must never be replayed as side-effectful work"); + assert_eq!(dead.pending_count(), 0); + assert_eq!(neighbor.pending_count(), 0); +} + +/// Queue admission must reject excess work promptly rather than retaining +/// unlimited messages that make an app with many sessions appear hung. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn bounded_per_session_queue_rejects_overload_and_drains_cleanly() { + let scheduler = DialogScheduler::new("stress-lab-queue-cap", 2); + let (first_started_tx, first_started_rx) = tokio::sync::oneshot::channel(); + let (release, hold) = tokio::sync::oneshot::channel::<()>(); + let finished = Arc::new(AtomicUsize::new(0)); + let initial_finished = Arc::clone(&finished); + + scheduler.enqueue(synthetic_message( + "blocking-first".into(), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + let _ = first_started_tx.send(()); + let _ = hold.await; + initial_finished.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("first maintenance job enqueued"); + tokio::time::timeout(TEST_DEADLINE, first_started_rx) + .await.expect("first job reaches its wait point") + .expect("first start signal"); + + for index in 0..2 { + let count = Arc::clone(&finished); + scheduler.enqueue(synthetic_message( + format!("queued-{index}"), ScheduledKind::Maintenance, + Box::new(move || Box::pin(async move { + count.fetch_add(1, Ordering::SeqCst); + Ok(ExecutionCompletion::Finished) + })), + )).await.expect("up to capacity pending jobs admitted"); + } + assert_eq!(scheduler.pending_count(), 2); + let rejected = scheduler.enqueue(synthetic_message( + "over-capacity".into(), ScheduledKind::Maintenance, + Box::new(|| Box::pin(async { Ok(ExecutionCompletion::Finished) })), + )).await.expect_err("the full per-session queue must reject immediately"); + assert!(rejected.contains("full"), "unexpected rejection: {rejected}"); + assert_eq!(scheduler.pending_count(), 2); + + release.send(()).expect("first job is still blocked"); + tokio::time::timeout(TEST_DEADLINE, async { + while finished.load(Ordering::SeqCst) != 3 { + tokio::task::yield_now().await; + } + }).await.expect("accepted queue drains after releasing the first job"); + assert_eq!(scheduler.pending_count(), 0); +} diff --git a/src-tauri/crates/agent-core/src/core/tools/impls/desktop/escape_hotkey.rs b/src-tauri/crates/agent-core/src/core/tools/impls/desktop/escape_hotkey.rs index 9d72d17bb3..ffe9ef0d9a 100644 --- a/src-tauri/crates/agent-core/src/core/tools/impls/desktop/escape_hotkey.rs +++ b/src-tauri/crates/agent-core/src/core/tools/impls/desktop/escape_hotkey.rs @@ -35,6 +35,24 @@ pub fn notify_expected_escape() { EXPECTED_ESCAPES.fetch_add(1, Ordering::Relaxed); } +/// Consume one expected synthetic Escape without underflow. A CAS loop +/// avoids deprecated fetch_update and works on the project's minimum Rust. +fn consume_expected_escape(counter: &AtomicU32) -> bool { + let mut remaining = counter.load(Ordering::SeqCst); + while remaining > 0 { + match counter.compare_exchange_weak( + remaining, + remaining - 1, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => return true, + Err(actual) => remaining = actual, + } + } + false +} + /// Enable or disable the abort action (the tap itself keeps running). pub fn set_enabled(enabled: bool) { ENABLED.store(enabled, Ordering::Relaxed); @@ -133,16 +151,7 @@ extern "C" fn escape_tap_callback( if is_synthetic { return event; } - if EXPECTED_ESCAPES - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| { - if n > 0 { - Some(n - 1) - } else { - None - } - }) - .is_ok() - { + if consume_expected_escape(&EXPECTED_ESCAPES) { return event; } @@ -191,3 +200,19 @@ fn run_event_tap() { CFRelease(tap); } } + +#[cfg(test)] +mod atomic_escape_tests { + use super::consume_expected_escape; + use std::sync::atomic::{AtomicU32, Ordering}; + + #[test] + fn expected_escapes_are_consumed_once() { + let count = AtomicU32::new(2); + assert!(consume_expected_escape(&count)); + assert!(consume_expected_escape(&count)); + assert!(!consume_expected_escape(&count)); + assert_eq!(count.load(Ordering::SeqCst), 0); + } +} + diff --git a/src-tauri/crates/agent-core/src/core/tools/traits.rs b/src-tauri/crates/agent-core/src/core/tools/traits.rs index 9998e395e8..1b354ea797 100644 --- a/src-tauri/crates/agent-core/src/core/tools/traits.rs +++ b/src-tauri/crates/agent-core/src/core/tools/traits.rs @@ -90,6 +90,8 @@ use super::params::ensure_object_schema; /// Each tool provides a name, description, JSON Schema parameters, /// and an async execute method. Tools are registered in a [`ToolRegistry`] /// and invoked by the agent loop when the LLM requests a tool call. +// async_trait's boxed Future is already must_use; scope this lint to this trait. +#[allow(clippy::double_must_use)] #[async_trait] pub trait Tool: Send + Sync { /// Tool name used in function calls (e.g., "read_file", "exec"). diff --git a/src-tauri/crates/agent-core/src/core/turn_executor/types.rs b/src-tauri/crates/agent-core/src/core/turn_executor/types.rs index 137c361caa..2bb5380e56 100644 --- a/src-tauri/crates/agent-core/src/core/turn_executor/types.rs +++ b/src-tauri/crates/agent-core/src/core/turn_executor/types.rs @@ -31,6 +31,7 @@ pub enum PermissionVerdict { /// `execute_turn` calls this when a tool has an `Ask` verdict in the policy. /// Implementations handle the actual prompt UI (a frontend dialog driven by /// a `permission:request` event delivered over the Tauri IPC Channel). +#[allow(clippy::double_must_use)] #[async_trait] pub trait PermissionProvider: Send + Sync { /// Check if a tool has been granted "always allow" for the current session. @@ -53,6 +54,7 @@ pub trait PermissionProvider: Send + Sync { /// before `execute_turn`, then perform a zero-wait collect before each provider /// request. Implementations must return immediately when their work is not yet /// ready. +#[allow(clippy::double_must_use)] #[async_trait] pub trait TurnIterationHook: Send + Sync { async fn before_llm_iteration( @@ -210,6 +212,8 @@ pub struct ToolHookIntervention { /// /// Implementors handle streaming deltas, tool call/result events, /// and persistence. Each agent type provides its own implementation. +// async_trait yields a must_use boxed Future; guard only this callback trait. +#[allow(clippy::double_must_use)] #[async_trait] #[allow(unused_variables, clippy::too_many_arguments)] // Event implementations share this callback contract across providers; keep diff --git a/src-tauri/crates/agent-core/src/integrations/channels/traits.rs b/src-tauri/crates/agent-core/src/integrations/channels/traits.rs index d6426f26e6..1368409ce4 100644 --- a/src-tauri/crates/agent-core/src/integrations/channels/traits.rs +++ b/src-tauri/crates/agent-core/src/integrations/channels/traits.rs @@ -9,6 +9,8 @@ use crate::bus::{InboundMessage, OutboundMessage}; /// /// Each channel implementation handles connecting to an external service /// (Telegram, Discord, etc.), receiving messages, and sending responses. +// async_trait's boxed Future is already must_use; keep the lint exception local. +#[allow(clippy::double_must_use)] #[async_trait] pub trait Channel: Send + Sync { /// Channel name including account ID (e.g., "telegram:default", "discord:work"). diff --git a/src-tauri/crates/agent-core/src/integrations/gateway/service.rs b/src-tauri/crates/agent-core/src/integrations/gateway/service.rs index 3394135582..2385590f48 100644 --- a/src-tauri/crates/agent-core/src/integrations/gateway/service.rs +++ b/src-tauri/crates/agent-core/src/integrations/gateway/service.rs @@ -23,6 +23,8 @@ use super::workers::{spawn_inbound_processor, spawn_outbound_dispatcher}; /// /// The gateway delegates message processing to this handler, which is implemented /// by the agent layer. This decouples the gateway from any specific agent. +// async_trait's generated boxed Future already has must_use semantics. +#[allow(clippy::double_must_use)] #[async_trait::async_trait] pub trait InboundMessageHandler: Send + Sync + 'static { async fn handle_message(&self, msg: InboundMessage) -> Result, String>; diff --git a/src-tauri/crates/agent-core/src/specialization/memory/commands.rs b/src-tauri/crates/agent-core/src/specialization/memory/commands.rs index 4b1114158f..fc29407796 100644 --- a/src-tauri/crates/agent-core/src/specialization/memory/commands.rs +++ b/src-tauri/crates/agent-core/src/specialization/memory/commands.rs @@ -674,25 +674,19 @@ pub async fn workspace_memory_status(workspace: String) -> Result { - consolidation_lock::rollback(ws, prior); - false - } - Ok(None) => true, + // Only probe the OS lease; status requests must never modify the + // persistent consolidation timestamp or roll back another writer. + let lock_held = match consolidation_lock::is_lease_held(ws) { + Ok(held) => held, Err(err) => { tracing::warn!( workspace = %ws.display(), error = %err, - "memory::workspace_memory_status: try_acquire failed; reporting lock_held=false but the actual lock state is unknown" + "memory::workspace_memory_status: lease probe failed; lock state unknown" ); - false + // Fail-closed for a UI status: unknown is not proof that the + // workspace can safely start another consolidation. + true } }; diff --git a/src-tauri/crates/agent-core/src/specialization/memory/embeddings/mod.rs b/src-tauri/crates/agent-core/src/specialization/memory/embeddings/mod.rs index c9e4547420..164c0132ac 100644 --- a/src-tauri/crates/agent-core/src/specialization/memory/embeddings/mod.rs +++ b/src-tauri/crates/agent-core/src/specialization/memory/embeddings/mod.rs @@ -29,6 +29,8 @@ pub struct EmbeddingResult { } /// Trait for embedding providers. +// async_trait generates a boxed Future already considered must_use. +#[allow(clippy::double_must_use)] #[async_trait] pub trait EmbeddingProvider: Send + Sync { /// Generate an embedding for a single text. diff --git a/src-tauri/crates/agent-core/src/specialization/memory/workspace_memory/lock.rs b/src-tauri/crates/agent-core/src/specialization/memory/workspace_memory/lock.rs index ebd3754bfb..4eca1a8c09 100644 --- a/src-tauri/crates/agent-core/src/specialization/memory/workspace_memory/lock.rs +++ b/src-tauri/crates/agent-core/src/specialization/memory/workspace_memory/lock.rs @@ -10,12 +10,17 @@ //! files), NOT the L3 learnings consolidation engine in //! `super::super::consolidation`. The two subsystems are unrelated. -use std::fs; +use std::fs::{self, File, OpenOptions}; use std::path::{Path, PathBuf}; +use std::time::Duration; use std::time::SystemTime; use tracing::warn; const LOCK_FILE: &str = ".consolidate-lock"; +// A distinct, permanent inode used for atomic OS-level advisory locking. +// Never unlink this while any process may hold it: unlinking permits another +// process to open a different inode and incorrectly become an owner. +const LEASE_FILE: &str = ".consolidate-lease"; /// Stale threshold: even if the PID is live, reclaim after this duration. const HOLDER_STALE_MS: u64 = 60 * 60 * 1000; // 1 hour @@ -61,6 +66,8 @@ pub struct ConsolidationLease { workspace: PathBuf, prior_mtime_ms: u64, committed: bool, + // fs2/FileExt OS lock: held until after Drop has committed/rolled back. + _lock_file: File, } impl ConsolidationLease { @@ -77,15 +84,57 @@ impl Drop for ConsolidationLease { } } +/// Report whether another thread/process owns the actual OS lease. +/// Unlike the historical PID probe, this does not touch the timestamp and +/// cannot accidentally roll back a concurrent consolidation. +pub fn is_lease_held(workspace: &Path) -> Result { + let memory = super::memory_dir(workspace); + if !memory.exists() { + return Ok(false); + } + let lock_file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(memory.join(LEASE_FILE)) + .map_err(|err| format!("open consolidation lease for status: {err}"))?; + match fs2::FileExt::try_lock_exclusive(&lock_file) { + Ok(()) => { + // The lease is available. Closing the file releases the lock. + drop(lock_file); + Ok(false) + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => Ok(true), + Err(err) => Err(format!("probe consolidation lease: {err}")), + } +} + /// Try to acquire the consolidation lock as an RAII lease. pub fn try_acquire_lease(workspace: &Path) -> Result, String> { - try_acquire(workspace).map(|prior| { - prior.map(|prior_mtime_ms| ConsolidationLease { - workspace: workspace.to_path_buf(), - prior_mtime_ms, - committed: false, - }) - }) + let memory = super::memory_dir(workspace); + fs::create_dir_all(&memory).map_err(|err| format!("mkdir failed: {err}"))?; + let lock_file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(memory.join(LEASE_FILE)) + .map_err(|err| format!("open consolidation lease: {err}"))?; + match fs2::FileExt::try_lock_exclusive(&lock_file) { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return Ok(None), + Err(err) => return Err(format!("acquire consolidation lease: {err}")), + } + // The legacy timestamp/PID protocol is read and updated ONLY after the + // exclusive OS lock is held, so threads sharing a PID cannot both win. + let acquired = try_acquire(workspace)?; + Ok(acquired.map(|prior_mtime_ms| ConsolidationLease { + workspace: workspace.to_path_buf(), + prior_mtime_ms, + committed: false, + _lock_file: lock_file, + })) } /// Try to acquire the consolidation lock. @@ -93,8 +142,9 @@ pub fn try_acquire_lease(workspace: &Path) -> Result, /// Returns `Ok(prior_mtime_ms)` on success, `Ok(None)` if blocked by /// another process, `Err` on I/O failure. /// -/// The lock is "acquired" by writing our PID and verifying we won. -pub fn try_acquire(workspace: &Path) -> Result, String> { +/// The PID/timestamp step is only valid while the RAII lease holds the OS +/// lock on the stable `.consolidate-lease` inode. Never call it alone. +fn try_acquire(workspace: &Path) -> Result, String> { let path = lock_path(workspace); let mem_dir = super::memory_dir(workspace); @@ -261,39 +311,16 @@ fn is_process_running(pid: u32) -> bool { } } -/// Set file mtime (cross-platform). +/// Restore the previous consolidation mtime on macOS, Windows and Linux. +/// `FileTimes` avoids the earlier Unix-only utimensat implementation. fn set_file_mtime(path: &Path, mtime_ms: u64) -> Result<(), String> { - let secs = (mtime_ms / 1000) as i64; - - #[cfg(unix)] - { - use std::ffi::CString; - let path_cstr = CString::new(path.to_string_lossy().as_bytes()) - .map_err(|e| format!("CString: {}", e))?; - let times = [ - libc::timespec { - tv_sec: secs, - tv_nsec: ((mtime_ms % 1000) * 1_000_000) as _, - }, - libc::timespec { - tv_sec: secs, - tv_nsec: ((mtime_ms % 1000) * 1_000_000) as _, - }, - ]; - let ret = unsafe { libc::utimensat(libc::AT_FDCWD, path_cstr.as_ptr(), times.as_ptr(), 0) }; - if ret != 0 { - return Err(format!( - "utimensat failed: {}", - std::io::Error::last_os_error() - )); - } - Ok(()) - } - #[cfg(not(unix))] - { - let _ = (path, secs); - Err("set_file_mtime not implemented on this platform".to_string()) - } + let timestamp = SystemTime::UNIX_EPOCH + Duration::from_millis(mtime_ms); + let file = OpenOptions::new() + .write(true) + .open(path) + .map_err(|err| format!("open mtime target: {err}"))?; + file.set_times(fs::FileTimes::new().set_modified(timestamp)) + .map_err(|err| format!("set mtime: {err}")) } #[cfg(test)] @@ -317,24 +344,22 @@ mod tests { #[test] fn test_acquire_fresh() { let tmp = TempDir::new().unwrap(); - let result = try_acquire(tmp.path()); - assert!(result.is_ok()); - let prior = result.unwrap(); - assert_eq!(prior, Some(0)); // No prior lock + let lease = try_acquire_lease(tmp.path()).unwrap(); + assert!(lease.is_some()); + assert!(read_last_consolidated_at(tmp.path()) > 0); + drop(lease); + assert_eq!(read_last_consolidated_at(tmp.path()), 0); } #[test] fn test_acquire_blocks_when_live() { let tmp = TempDir::new().unwrap(); - - // First acquire - let r1 = try_acquire(tmp.path()).unwrap(); - assert!(r1.is_some()); - - // Second acquire from same process is blocked because the lock is - // fresh and our PID is live. - let r2 = try_acquire(tmp.path()).unwrap(); - assert!(r2.is_none(), "should be blocked by live holder"); + let first = try_acquire_lease(tmp.path()).unwrap().unwrap(); + // OS advisory locking must also block within a single process, + // where both contenders have the same PID; this works on Windows. + let second = try_acquire_lease(tmp.path()).unwrap(); + assert!(second.is_none(), "should be blocked by OS lock holder"); + drop(first); } #[test] @@ -382,6 +407,104 @@ mod tests { assert!(read_last_consolidated_at(tmp.path()) > 0); } + #[test] + fn status_probe_does_not_modify_memory_timestamp() { + let tmp = TempDir::new().unwrap(); + assert!(!is_lease_held(tmp.path()).unwrap()); + assert_eq!(read_last_consolidated_at(tmp.path()), 0); + let lease = try_acquire_lease(tmp.path()).unwrap().unwrap(); + let stamp = read_last_consolidated_at(tmp.path()); + assert!(is_lease_held(tmp.path()).unwrap()); + assert_eq!(read_last_consolidated_at(tmp.path()), stamp); + lease.commit(); + assert!(!is_lease_held(tmp.path()).unwrap()); + assert_eq!(read_last_consolidated_at(tmp.path()), stamp); + } + + #[test] + fn lease_blocks_another_thread_in_same_process() { + let tmp = TempDir::new().unwrap(); + let first = try_acquire_lease(tmp.path()).unwrap().unwrap(); + assert!(try_acquire_lease(tmp.path()).unwrap().is_none()); + drop(first); + assert!(try_acquire_lease(tmp.path()).unwrap().is_some()); + } + + #[test] + fn simultaneous_acquirers_have_exactly_one_owner() { + use std::sync::{Arc, Barrier}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + let tmp = TempDir::new().unwrap(); + let path = tmp.path().to_owned(); + let start = Arc::new(Barrier::new(9)); + let hold = Arc::new(Barrier::new(9)); + let acquired = Arc::new(AtomicUsize::new(0)); + let mut threads = Vec::new(); + for _ in 0..8 { + let (p, s, h, count) = (path.clone(), start.clone(), hold.clone(), acquired.clone()); + threads.push(std::thread::spawn(move || { + s.wait(); + let lease = try_acquire_lease(&p).expect("OS lock"); + if lease.is_some() { + count.fetch_add(1, Ordering::SeqCst); + } + h.wait(); // Keep the winner alive until every contender checked. + drop(lease); + })); + } + start.wait(); + hold.wait(); + for thread in threads { thread.join().unwrap(); } + assert_eq!(acquired.load(Ordering::SeqCst), 1); + } + + #[test] + fn subprocess_cannot_steal_parent_lease() { + const CHILD_VAR: &str = "TP_DREAM_LOCK_CHILD_WORKSPACE"; + if let Some(workspace) = std::env::var_os(CHILD_VAR) { + assert!( + try_acquire_lease(Path::new(&workspace)).unwrap().is_none(), + "child process must not obtain the parent's OS lease" + ); + return; + } + + let tmp = TempDir::new().unwrap(); + let parent = try_acquire_lease(tmp.path()).unwrap().unwrap(); + let child = std::process::Command::new(std::env::current_exe().unwrap()) + .arg("subprocess_cannot_steal_parent_lease") + .arg("--nocapture") + .env(CHILD_VAR, tmp.path()) + .output() + .expect("launch child test binary"); + assert!( + child.status.success(), + "child lock test failed:\\n{}\\n{}", + String::from_utf8_lossy(&child.stdout), + String::from_utf8_lossy(&child.stderr) + ); + drop(parent); + } + + #[test] + fn rollback_restores_previous_timestamp_across_platforms() { + let tmp = TempDir::new().unwrap(); + record_consolidation(tmp.path()).unwrap(); + // Move the timestamp beyond the stale PID threshold so the next + // acquisition is eligible while retaining a nonzero prior mtime. + let previous = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 - HOLDER_STALE_MS - 60_000; + set_file_mtime(&lock_path(tmp.path()), previous).unwrap(); + let prior = read_last_consolidated_at(tmp.path()); + let lease = try_acquire_lease(tmp.path()).unwrap().unwrap(); + drop(lease); + let restored = read_last_consolidated_at(tmp.path()); + assert_eq!(restored, prior); + } + #[test] fn test_lock_path() { let workspace = Path::new("/home/user/workspace"); diff --git a/src-tauri/crates/app-paths/src/data_root.rs b/src-tauri/crates/app-paths/src/data_root.rs index 3ff5822c75..a641ef4744 100644 --- a/src-tauri/crates/app-paths/src/data_root.rs +++ b/src-tauri/crates/app-paths/src/data_root.rs @@ -1,4 +1,4 @@ -//! The `~/.orgii/` data root and the files and directories that live directly +//! The fork-local `~/.tp-agent-studio/` data root and the files and directories that live directly //! under it: databases, caches, settings, the OS Agent ("personal") family, //! sync stores, policy/MCP config, and the file-based session registry. @@ -6,17 +6,17 @@ use std::path::{Path, PathBuf}; use crate::home::home_dir; -/// Application data root: `~/.orgii/`. +/// Independent fork data root: `~/.tp-agent-studio/` by default. /// -/// Test override: setting `ORGII_HOME` redirects every path under the data -/// root to that directory. Production code must never set it. -/// -/// Fallback: `$TMPDIR/.orgii/` (only reached if `dirs::home_dir()` fails). +/// `ORGII_HOME` is still the internal compatibility override for child tools, +/// tests and the bootstrap layer. Never point it to the upstream `~/.orgii/`. +/// Fork-specific Tauri startup selects distinct primary/dev/numbered roots. +/// Fallback: `$TMPDIR/.tp-agent-studio/` when user home is unavailable. pub fn orgii_root() -> PathBuf { if let Ok(override_path) = std::env::var("ORGII_HOME") { return PathBuf::from(override_path); } - home_dir().join(".orgii") + home_dir().join(".tp-agent-studio") } /// Provider keys file: `~/.orgii/credentials.json`. diff --git a/src-tauri/crates/app-window/src/dock_icon.rs b/src-tauri/crates/app-window/src/dock_icon.rs index de62baf3ec..0f7b5a7948 100644 --- a/src-tauri/crates/app-window/src/dock_icon.rs +++ b/src-tauri/crates/app-window/src/dock_icon.rs @@ -14,11 +14,11 @@ use tauri::AppHandle; /// `src/config/settingsSchema/registry/general.ts`. pub const DOCK_ICON_SETTING_KEY: &str = "general.dockIcon"; -/// Bundled 512×512 PNGs rendered from `src/assets/appIcons/*.svg`. +/// Bundled 512×512 PNGs (the default is the fork-owned two-tone icon). /// All variants omit the circle behind the mark; light keeps a soft inset /// edge so the white tile still reads on a light Dock. const DEV_ICON_PNG: &[u8] = include_bytes!("../../../icons/dev/icon.png"); -const DEV_IDENTIFIER: &str = "org2ai.org2.dev"; +const DEV_IDENTIFIER: &str = "com.theerasakping.tpagentstudio.dev"; const DARK_ICON_PNG: &[u8] = include_bytes!("../../../icons/dock/dark.png"); const LIGHT_ICON_PNG: &[u8] = include_bytes!("../../../icons/dock/light.png"); @@ -64,7 +64,7 @@ impl DockIconVariant { } } -/// Dev shares settings with the installed app, so its visual identity must +/// Dev has its own isolated settings store; its visual identity must /// win over every stored/live preference without changing that preference. fn process_icon_png(identifier: &str, variant: DockIconVariant) -> &'static [u8] { if identifier == DEV_IDENTIFIER { @@ -233,10 +233,10 @@ mod tests { DockIconVariant::Rainbow, ] { assert_eq!( - super::process_icon_png("org2ai.org2.dev", variant), + super::process_icon_png("com.theerasakping.tpagentstudio.dev", variant), super::DEV_ICON_PNG ); - for identifier in ["org2ai.org2", "org2ai.org2.instance2"] { + for identifier in ["com.theerasakping.tpagentstudio", "com.theerasakping.tpagentstudio.instance2"] { assert_eq!(super::process_icon_png(identifier, variant), variant.png()); } } diff --git a/src-tauri/crates/database/src/db/writer/fault_injection_tests.rs b/src-tauri/crates/database/src/db/writer/fault_injection_tests.rs new file mode 100644 index 0000000000..1523919b9c --- /dev/null +++ b/src-tauri/crates/database/src/db/writer/fault_injection_tests.rs @@ -0,0 +1,153 @@ +//! Isolated SQLite contention simulations using the production WAL writer +//! serializer and BEGIN IMMEDIATE helper. Never opens the user's sessions.db. +use super::{begin_immediate, with_sessions_writer}; +use rusqlite::{params, Connection, Result as SqliteResult}; +use std::sync::{mpsc, Arc, Barrier}; +use std::thread; +use std::time::Duration; + +const DEADLINE: Duration = Duration::from_secs(20); + +fn isolated_db() -> (tempfile::TempDir, std::path::PathBuf) { + let sandbox = tempfile::tempdir().expect("isolated tempdir"); + let path = sandbox.path().join("fault-lab.sqlite3"); + let conn = Connection::open(&path).expect("open isolated database"); + conn.execute_batch( + "PRAGMA journal_mode = WAL; + CREATE TABLE IF NOT EXISTS synthetic_turn_results( + session_no INTEGER NOT NULL, + turn_no INTEGER NOT NULL, + PRIMARY KEY(session_no, turn_no) + );", + ).expect("initialize WAL and isolated fixture schema"); + drop(conn); + (sandbox, path) +} + +fn concurrent_wal_profile(worker_count: usize, rows_per_worker: usize) { + // Files are scoped to a unique TempDir. These tests must never point + // get_connection() at an actual user sessions database or reset its pool. + let (_sandbox, path) = isolated_db(); + let gate = Arc::new(Barrier::new(worker_count + 1)); + let (notify, finished) = mpsc::channel::<(usize, SqliteResult<()>)>(); + let mut threads = Vec::with_capacity(worker_count); + + for worker in 0..worker_count { + let writer_path = path.clone(); + let barrier = Arc::clone(&gate); + let result_tx = notify.clone(); + threads.push(thread::spawn(move || { + barrier.wait(); + let result = (|| -> SqliteResult<()> { + let conn = Connection::open(&writer_path)?; + // With the Rust writer mutex, concurrent in-process writes + // must serialize without ever exhausting this backstop. + conn.busy_timeout(Duration::from_millis(500))?; + with_sessions_writer(|| { + let tx = begin_immediate(&conn)?; + for turn in 0..rows_per_worker { + tx.execute( + "INSERT INTO synthetic_turn_results(session_no, turn_no) + VALUES (?1, ?2)", + params![worker as i64, turn as i64], + )?; + } + tx.commit() + }) + })(); + let _ = result_tx.send((worker, result)); + })); + } + drop(notify); + gate.wait(); + + // A hung writer is a bounded test failure rather than an indefinitely + // running CI job. No detached cleanup actor or retry loop is introduced. + for _ in 0..worker_count { + let (worker, result) = finished.recv_timeout(DEADLINE) + .expect("concurrent writer did not finish before bounded deadline"); + result.unwrap_or_else(|error| { + panic!("SQLite write failed in synthetic Session {worker}: {error}") + }); + } + for handle in threads { + handle.join().expect("synthetic writer thread panicked"); + } + + let conn = Connection::open(&path).expect("verify isolated database"); + let recorded: i64 = conn.query_row( + "SELECT COUNT(*) FROM synthetic_turn_results", [], |row| row.get(0), + ).expect("read committed row count"); + assert_eq!(recorded, (worker_count * rows_per_worker) as i64, + "lost writes or duplicate turns under concurrent WAL writers"); + let mut statement = conn.prepare( + "SELECT session_no, COUNT(*) FROM synthetic_turn_results GROUP BY session_no", + ).expect("prepare per-session count"); + let per_session = statement.query_map([], |row| { + Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)) + }).expect("group query"); + let mut seen = 0usize; + for item in per_session { + let (_, count) = item.expect("read per-session result"); + assert_eq!(count, rows_per_worker as i64); + seen += 1; + } + assert_eq!(seen, worker_count); + let integrity: String = conn.query_row("PRAGMA integrity_check", [], |row| row.get(0)) + .expect("verify SQLite file integrity"); + assert_eq!(integrity, "ok"); +} + +#[test] +fn five_simultaneous_wal_writers_are_serialized() { + concurrent_wal_profile(5, 12); +} + +#[test] +fn twenty_simultaneous_wal_writers_are_serialized() { + concurrent_wal_profile(20, 8); +} + +#[test] +fn fifty_simultaneous_wal_writers_are_serialized() { + concurrent_wal_profile(50, 6); +} + +/// In WAL mode, an unrelated reader must be able to observe the last +/// committed snapshot while one writer owns an uncommitted transaction. +#[test] +fn reader_is_not_blocked_by_an_in_process_writer_transaction() { + let (_sandbox, path) = isolated_db(); + let (entered_tx, entered_rx) = mpsc::sync_channel::<()>(1); + let (release_tx, release_rx) = mpsc::sync_channel::<()>(1); + let writer_path = path.clone(); + let writer = thread::spawn(move || -> SqliteResult<()> { + let conn = Connection::open(&writer_path)?; + with_sessions_writer(|| { + let tx = begin_immediate(&conn)?; + tx.execute( + "INSERT INTO synthetic_turn_results(session_no, turn_no) VALUES(1, 1)", [], + )?; + let _ = entered_tx.send(()); + release_rx.recv_timeout(Duration::from_secs(5)) + .expect("test reader must release the writer"); + tx.commit() + }) + }); + entered_rx.recv_timeout(Duration::from_secs(5)) + .expect("writer must reach held transaction"); + + let reader = Connection::open(&path).expect("open independent SQLite reader"); + reader.busy_timeout(Duration::from_millis(500)).expect("set reader timeout"); + let before: i64 = reader.query_row( + "SELECT COUNT(*) FROM synthetic_turn_results", [], |row| row.get(0), + ).expect("WAL reader must see committed state while writer has BEGIN IMMEDIATE"); + assert_eq!(before, 0, "uncommitted writes must not leak to another reader"); + + release_tx.send(()).expect("release held writer transaction"); + writer.join().expect("writer thread completes").expect("commit succeeds"); + let after: i64 = reader.query_row( + "SELECT COUNT(*) FROM synthetic_turn_results", [], |row| row.get(0), + ).expect("reader can see committed result"); + assert_eq!(after, 1); +} diff --git a/src-tauri/crates/project-management/src/sync/adapter.rs b/src-tauri/crates/project-management/src/sync/adapter.rs index 5281386ba6..c0d7932670 100644 --- a/src-tauri/crates/project-management/src/sync/adapter.rs +++ b/src-tauri/crates/project-management/src/sync/adapter.rs @@ -262,6 +262,10 @@ pub struct SyncStatusReport { /// Every adapter is a unit-struct implementing this trait. Registration /// goes through [`super::adapters::registry`] at boot so commands can /// look up adapters by ID without hard-coding any. +// async_trait materializes async methods as boxed Future-returning functions. +// Current Clippy sees the macro's #[must_use] on those already must-use Futures. +// Scope the compatibility waiver to this trait only, not the entire crate. +#[allow(clippy::double_must_use)] #[async_trait] pub trait SyncAdapter: Send + Sync { /// Stable identifier (snake_case). Used as the registry key and diff --git a/src-tauri/crates/search/src/code/commands/text_search.rs b/src-tauri/crates/search/src/code/commands/text_search.rs index 4bad8bc62b..3886b2e204 100644 --- a/src-tauri/crates/search/src/code/commands/text_search.rs +++ b/src-tauri/crates/search/src/code/commands/text_search.rs @@ -23,12 +23,24 @@ struct SearchBudget { struct Permit; impl Permit { fn acquire() -> Result { - WORKERS - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| { - (n < 4).then_some(n + 1) - }) - .map(|_| Self) - .map_err(|_| "Search workers are busy; retry after pending searches finish".into()) + // Portable to the project's minimum Rust version and current stable: + // newer toolchains deprecate fetch_update in favor of try_update. + // A CAS loop retains the original 4-worker admission bound. + let mut workers = WORKERS.load(Ordering::Acquire); + loop { + if workers >= 4 { + return Err("Search workers are busy; retry after pending searches finish".into()); + } + match WORKERS.compare_exchange_weak( + workers, + workers + 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return Ok(Self), + Err(actual) => workers = actual, + } + } } } impl Drop for Permit { diff --git a/src-tauri/crates/system-services/src/tray.rs b/src-tauri/crates/system-services/src/tray.rs index a13fce8e8c..8ec833e379 100644 --- a/src-tauri/crates/system-services/src/tray.rs +++ b/src-tauri/crates/system-services/src/tray.rs @@ -220,7 +220,7 @@ pub fn setup_tray(app: &AppHandle) -> Result<(), Box> { app, &[], "Open App", - "Quit ORG2", + "Quit TP Agent Studio", "Show Kanban", "Show Runtime", )?; diff --git a/src-tauri/crates/terminal/src/agent_tool/mod.rs b/src-tauri/crates/terminal/src/agent_tool/mod.rs index d39af4142b..5a0717cba1 100644 --- a/src-tauri/crates/terminal/src/agent_tool/mod.rs +++ b/src-tauri/crates/terminal/src/agent_tool/mod.rs @@ -37,6 +37,55 @@ use crate::pty_commands::pty::{encode_pty_output_frame, PtySession, PtySessionId use crate::pty_commands::shell_integration; use crate::pty_commands::shells::ShellKind; +#[cfg(test)] +mod atomic_sub_tests { + use super::saturating_atomic_sub; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn subtract_is_saturating_and_returns_previous_counter() { + let counter = AtomicUsize::new(100); + assert_eq!(saturating_atomic_sub(&counter, 40), 100); + assert_eq!(counter.load(Ordering::Relaxed), 60); + assert_eq!(saturating_atomic_sub(&counter, 1_000), 60); + assert_eq!(counter.load(Ordering::Relaxed), 0); + } + + #[test] + fn simultaneous_subtractions_never_underflow() { + use std::sync::Arc; + let counter = Arc::new(AtomicUsize::new(100)); + let mut joins = Vec::new(); + for _ in 0..16 { + let counter = counter.clone(); + joins.push(std::thread::spawn(move || { + saturating_atomic_sub(&counter, 30); + })); + } + for join in joins { + join.join().expect("atomic decrement thread"); + } + assert_eq!(counter.load(Ordering::Relaxed), 0); + } +} + +/// Atomically subtract acknowledged/dropped output bytes without underflow. +/// A CAS loop keeps this compatible with the project's declared Rust MSRV. +pub(crate) fn saturating_atomic_sub(counter: &AtomicUsize, bytes: usize) -> usize { + let mut before = counter.load(Ordering::Relaxed); + loop { + match counter.compare_exchange_weak( + before, + before.saturating_sub(bytes), + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(previous) => return previous, + Err(actual) => before = actual, + } + } +} + // ============================================ // Constants // ============================================ @@ -781,11 +830,7 @@ pub async fn create_session(params: CreateSessionParams) -> Result bool { if identifier != DEV_IDENTIFIER @@ -72,19 +72,25 @@ mod tests { let bare = Path::new("/build/debug/org2"); assert!(needs_named_executable(DEV_IDENTIFIER, bare)); for identifier in [ - "org2ai.org2", - "org2ai.org2.instance2", - "org2ai.org2.e2e.instance2", + "com.theerasakping.tpagentstudio", + "com.theerasakping.tpagentstudio.instance2", + "com.theerasakping.tpagentstudio.e2e.instance2", ] { assert!(!needs_named_executable(identifier, bare)); } assert!(!needs_named_executable( + DEV_IDENTIFIER, + Path::new("/build/debug/TP Agent Studio Dev") + )); + // The upstream process name is NOT the fork's dev process name. + // Otherwise a renamed fork could keep showing an upstream Dock title. + assert!(needs_named_executable( DEV_IDENTIFIER, Path::new("/build/debug/ORG2 Dev") )); assert!(!needs_named_executable( DEV_IDENTIFIER, - Path::new("/build/ORG2 Dev.app/Contents/MacOS/org2") + Path::new("/build/TP Agent Studio Dev.app/Contents/MacOS/org2") )); } diff --git a/src-tauri/src/app/startup_recovery.rs b/src-tauri/src/app/startup_recovery.rs index 693f03672c..27bad457af 100644 --- a/src-tauri/src/app/startup_recovery.rs +++ b/src-tauri/src/app/startup_recovery.rs @@ -26,25 +26,54 @@ pub(crate) fn run_persistence_startup_recovery() -> Result Result<(), String> { + if FORK_UPDATER_READY { + Ok(()) + } else { + Err("Updates are disabled in this independent fork until its own signed release channel is configured".to_string()) + } +} + const STABLE_MANIFEST_URL: &str = - "https://github.com/org2AI/ORG2/releases/latest/download/latest.json"; + "https://github.com/TheerasakPing/ORG2/releases/latest/download/latest.json"; /// Beta channel: rolling `updater` release whose `beta.json` is overwritten /// by every release (stable and beta) in `.github/workflows/release.yaml`, /// so it always points at the newest build of either kind. const BETA_MANIFEST_URL: &str = - "https://github.com/org2AI/ORG2/releases/download/updater/beta.json"; + "https://github.com/TheerasakPing/ORG2/releases/download/updater/beta.json"; -const PRODUCTION_MACOS_APP_PATH: &str = "/Applications/ORG2.app"; -const PRODUCTION_BUNDLE_IDENTIFIER: &str = "org2ai.org2"; +const PRODUCTION_MACOS_APP_PATH: &str = "/Applications/TP Agent Studio.app"; +const PRODUCTION_BUNDLE_IDENTIFIER: &str = "com.theerasakping.tpagentstudio"; static SEPARATE_INSTALL_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); @@ -76,6 +88,9 @@ fn compiled_build_kind() -> AppBuildKind { } fn install_strategy(kind: AppBuildKind) -> AppUpdateInstallStrategy { + if !FORK_UPDATER_READY { + return AppUpdateInstallStrategy::Unavailable; + } match kind { AppBuildKind::Release => AppUpdateInstallStrategy::InPlace, AppBuildKind::Local if cfg!(target_os = "macos") => { @@ -157,6 +172,7 @@ pub async fn check_app_update( channel: UpdateChannel, timeout_ms: Option, ) -> Result, String> { + require_fork_updater_ready()?; let endpoint = Url::parse(channel.manifest_url()).map_err(|err| err.to_string())?; let mut builder = webview @@ -394,17 +410,16 @@ mod tests { } #[test] - fn release_builds_update_in_place_and_local_builds_never_guess() { + fn independent_fork_disables_all_upstream_update_strategies() { assert_eq!( install_strategy(AppBuildKind::Release), - AppUpdateInstallStrategy::InPlace + AppUpdateInstallStrategy::Unavailable ); - let expected_local = if cfg!(target_os = "macos") { - AppUpdateInstallStrategy::SeparateMacosApplication - } else { + assert_eq!( + install_strategy(AppBuildKind::Local), AppUpdateInstallStrategy::Unavailable - }; - assert_eq!(install_strategy(AppBuildKind::Local), expected_local); + ); + assert!(require_fork_updater_ready().is_err()); } #[test] diff --git a/src-tauri/src/market_connection/native_admission.rs b/src-tauri/src/market_connection/native_admission.rs index 557c0d27da..5cdd4f270b 100644 --- a/src-tauri/src/market_connection/native_admission.rs +++ b/src-tauri/src/market_connection/native_admission.rs @@ -15,6 +15,7 @@ const MAX_WAITERS: usize = 128; const FRESHNESS: Duration = Duration::from_secs(30); const WAIT_BUDGET: Duration = Duration::from_secs(90); +#[allow(clippy::double_must_use)] #[async_trait] pub(super) trait CapacitySource: Send + Sync { async fn buyer_capacity(&self, metadata: &ConnectionMetadata) -> Result; diff --git a/src-tauri/src/market_connection/native_compatibility.rs b/src-tauri/src/market_connection/native_compatibility.rs index 89be12fb55..be1f7a2308 100644 --- a/src-tauri/src/market_connection/native_compatibility.rs +++ b/src-tauri/src/market_connection/native_compatibility.rs @@ -257,10 +257,23 @@ fn inspect(bundle: &Path, agent: &str) -> Result { // digest: bindings and model caches live in this same process. A new local // generation avoids reading hundreds of MB from executable/App archives. static GENERATION: AtomicU64 = AtomicU64::new(1); - let generation = GENERATION - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_add(1)) - .map_err(|_| "Native App generation exhausted")? - .to_string(); + // CAS loop is supported by our Rust MSRV and avoids both the deprecated + // fetch_update API and the new try_update API (stable only since 1.95). + let mut observed = GENERATION.load(Ordering::Relaxed); + let generation = loop { + let next = observed + .checked_add(1) + .ok_or("Native App generation exhausted")?; + match GENERATION.compare_exchange_weak( + observed, + next, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(previous) => break previous.to_string(), + Err(current) => observed = current, + } + }; #[cfg(not(any(target_os = "macos", test)))] let _ = (&version, &generation); let resolved = ResolvedNativeClient { diff --git a/src-tauri/src/runtime_instance.rs b/src-tauri/src/runtime_instance.rs index e1314bce51..f573761f79 100644 --- a/src-tauri/src/runtime_instance.rs +++ b/src-tauri/src/runtime_instance.rs @@ -10,6 +10,18 @@ use std::path::{Path, PathBuf}; const PRIMARY_IDE_SERVER_PORT: u16 = 13_847; const PRIMARY_CLI_PROXY_PORT: u16 = 17_888; + +// Independent fork: keep all TCP defaults outside upstream's 13847..13946 +// and 17888..17987 instance ranges (including the reserved dev slot). +const FORK_IDENTIFIER: &str = "com.theerasakping.tpagentstudio"; +const FORK_DEV_IDENTIFIER: &str = "com.theerasakping.tpagentstudio.dev"; +const FORK_IDE_SERVER_PORT: u16 = 23_847; +const FORK_CLI_PROXY_PORT: u16 = 27_888; +const FORK_INSTANCE_ID_BIAS: u16 = 1000; +const FORK_INSTANCE_IDENTIFIER_PREFIXES: &[&str] = &[ + "com.theerasakping.tpagentstudio.instance", + "com.theerasakping.tpagentstudio.e2e.instance", +]; // Dedicated dev slot, outside the numbered bundle range (2..=99). const DEV_INSTANCE_ID: u16 = 100; const INSTANCE_IDENTIFIER_PREFIXES: &[&str] = &["org2ai.org2.instance", "org2ai.org2.e2e.instance"]; @@ -22,7 +34,48 @@ pub(crate) struct RuntimeInstanceProfile { } impl RuntimeInstanceProfile { + pub(crate) fn is_fork(self) -> bool { + self.instance_id > FORK_INSTANCE_ID_BIAS + } + + fn for_fork_instance(instance_id: u16) -> Self { + Self { + instance_id: FORK_INSTANCE_ID_BIAS + instance_id, + ide_server_port: FORK_IDE_SERVER_PORT + instance_id - 1, + cli_proxy_port: FORK_CLI_PROXY_PORT + instance_id - 1, + } + } + + pub(crate) fn validate_explicit_home(self, user_home: &Path, configured: &Path) -> Result<(), &'static str> { + if !self.is_fork() { + return Ok(()); + } + if !configured.is_absolute() || configured.as_os_str().is_empty() { + return Err("Fork data root must be an absolute path"); + } + let upstream = user_home.join(".orgii"); + // Normalize '..' even when a directory has not been created yet and + // resolve the nearest existing ancestor. A plain canonicalize() + // fallback on the raw path let non-existent paths hide traversal or + // parent symlinks into the upstream profile. + let configured_real = resolve_profile_candidate(configured)?; + let upstream_real = resolve_profile_candidate(&upstream)?; + if profile_path_contains(&upstream_real, &configured_real) { + return Err("Fork data root must not point to the upstream ORG2 profile"); + } + Ok(()) + } + pub(crate) fn from_identifier(identifier: &str) -> Self { + if identifier == FORK_IDENTIFIER { + return Self::for_fork_instance(1); + } + if identifier == FORK_DEV_IDENTIFIER { + return Self::for_fork_instance(DEV_INSTANCE_ID); + } + if let Some(instance_id) = parse_fork_instance_id(identifier) { + return Self::for_fork_instance(instance_id); + } let instance_id = if identifier == "org2ai.org2.dev" { DEV_INSTANCE_ID } else { @@ -39,6 +92,15 @@ impl RuntimeInstanceProfile { /// Primary and dev intentionally share `~/.orgii`, including sessions.db /// and its attachments/replays, by returning `None`. pub(crate) fn default_orgii_home(self, user_home: &Path) -> Option { + if self.is_fork() { + let slot = self.instance_id - FORK_INSTANCE_ID_BIAS; + let folder = match slot { + 1 => ".tp-agent-studio".to_string(), + DEV_INSTANCE_ID => ".tp-agent-studio-dev".to_string(), + id => format!(".tp-agent-studio-instance{id}"), + }; + return Some(user_home.join(folder)); + } if self.instance_id == DEV_INSTANCE_ID { return None; } @@ -52,11 +114,87 @@ impl RuntimeInstanceProfile { self, resolved_orgii_home: &Path, ) -> Option { - (self.instance_id > 1 && self.instance_id != DEV_INSTANCE_ID) + (self.is_fork() || (self.instance_id > 1 && self.instance_id != DEV_INSTANCE_ID)) .then(|| resolved_orgii_home.join("external-history-home")) } } +/// Normalize a full, absolute profile path before consulting the filesystem. +/// No filesystem modifications are made: existing symlinks are resolved, while +/// missing trailing components retain their identity under an existing parent. +/// This guards accidental overlaps; it is not a symlink-race sandbox. +fn resolve_profile_candidate(path: &Path) -> Result { + use std::path::Component; + if !path.is_absolute() { + return Err("Profile path must be absolute"); + } + let mut normalized = PathBuf::new(); + for component in path.components() { + match component { + Component::Prefix(prefix) => normalized.push(prefix.as_os_str()), + Component::RootDir => normalized.push(component.as_os_str()), + Component::CurDir => {} + Component::Normal(part) => normalized.push(part), + Component::ParentDir => { + if !normalized.pop() { + return Err("Profile path escapes the filesystem root"); + } + } + } + } + + let mut parent = normalized; + let mut missing = Vec::new(); + loop { + match std::fs::symlink_metadata(&parent) { + Ok(_) => { + let mut real = parent + .canonicalize() + .map_err(|_| "Cannot resolve a symlink in the profile path")?; + for suffix in missing.iter().rev() { + real.push(suffix); + } + return Ok(real); + } + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + let Some(name) = parent.file_name() else { + return Err("Cannot resolve the profile path root"); + }; + missing.push(name.to_os_string()); + if !parent.pop() { + return Err("Cannot resolve the profile path parent"); + } + } + Err(_) => return Err("Cannot inspect the profile path"), + } + } +} + +/// Windows paths are case-insensitive even when a not-yet-created child +/// does not exist for std::fs::canonicalize to normalize its final segment. +fn profile_path_contains(parent: &Path, candidate: &Path) -> bool { + #[cfg(windows)] + { + let p = parent.to_string_lossy().to_ascii_lowercase(); + let c = candidate.to_string_lossy().to_ascii_lowercase(); + let sep = std::path::MAIN_SEPARATOR; + c == p || c.starts_with(&format!("{p}{sep}")) + } + #[cfg(not(windows))] + { + candidate.starts_with(parent) + } +} + +fn parse_fork_instance_id(identifier: &str) -> Option { + FORK_INSTANCE_IDENTIFIER_PREFIXES + .iter() + .find_map(|prefix| identifier.strip_prefix(prefix))? + .parse::() + .ok() + .filter(|id| (2..=99).contains(id)) +} + fn parse_instance_id(identifier: &str) -> Option { INSTANCE_IDENTIFIER_PREFIXES .iter() @@ -70,6 +208,82 @@ fn parse_instance_id(identifier: &str) -> Option { mod tests { use super::*; + #[test] + fn rejects_missing_suffix_and_parent_traversal_into_upstream_data() { + let tmp = tempfile::tempdir().unwrap(); + let user_home = tmp.path(); + let upstream = user_home.join(".orgii"); + std::fs::create_dir_all(&upstream).unwrap(); + + let fork = RuntimeInstanceProfile::from_identifier(FORK_IDENTIFIER); + assert!(fork.validate_explicit_home(user_home, &upstream).is_err()); + assert!(fork + .validate_explicit_home(user_home, &user_home.join("scratch/../.orgii/secrets")) + .is_err()); + assert!(fork + .validate_explicit_home(user_home, &upstream.join("not-yet-created/sessions.db")) + .is_err()); + assert!(fork + .validate_explicit_home(user_home, &user_home.join(".orgii-suffix")) + .is_ok()); + assert!(fork + .validate_explicit_home(user_home, &user_home.join(".tp-agent-studio")) + .is_ok()); + } + + #[cfg(unix)] + #[test] + fn rejects_symlinked_parent_into_upstream_profile() { + use std::os::unix::fs::symlink; + let tmp = tempfile::tempdir().unwrap(); + let upstream = tmp.path().join(".orgii"); + std::fs::create_dir_all(&upstream).unwrap(); + let link = tmp.path().join("alias"); + symlink(&upstream, &link).unwrap(); + let fork = RuntimeInstanceProfile::from_identifier(FORK_IDENTIFIER); + assert!(fork.validate_explicit_home(tmp.path(), &link).is_err()); + assert!(fork + .validate_explicit_home(tmp.path(), &link.join("new-profile")) + .is_err()); + } + + #[test] + fn independent_fork_uses_distinct_ports_and_profiles() { + let upstream = RuntimeInstanceProfile::from_identifier("org2ai.org2"); + let fork = RuntimeInstanceProfile::from_identifier(FORK_IDENTIFIER); + let fork_dev = RuntimeInstanceProfile::from_identifier(FORK_DEV_IDENTIFIER); + assert!(fork.is_fork()); + assert!(fork_dev.is_fork()); + assert_ne!(fork.ide_server_port, upstream.ide_server_port); + assert_ne!(fork.cli_proxy_port, upstream.cli_proxy_port); + assert_eq!(fork.ide_server_port, 23_847); + assert_eq!(fork.cli_proxy_port, 27_888); + assert_eq!(fork.default_orgii_home(Path::new("/home/test")), Some(PathBuf::from("/home/test/.tp-agent-studio"))); + assert_eq!(fork_dev.default_orgii_home(Path::new("/home/test")), Some(PathBuf::from("/home/test/.tp-agent-studio-dev"))); + assert_ne!(fork.ide_server_port, fork_dev.ide_server_port); + assert_eq!(fork.default_external_history_home(Path::new("/home/test/.tp-agent-studio")), Some(PathBuf::from("/home/test/.tp-agent-studio/external-history-home"))); + let temp_home = tempfile::tempdir().unwrap(); + assert!(fork.validate_explicit_home(temp_home.path(), &temp_home.path().join(".orgii")).is_err()); + assert!(fork.validate_explicit_home(temp_home.path(), Path::new("relative")).is_err()); + assert!(fork.validate_explicit_home(temp_home.path(), &temp_home.path().join(".tp-agent-studio-e2e")).is_ok()); + } + + #[test] + fn fork_numbered_instances_preserve_separate_profiles() { + let primary = RuntimeInstanceProfile::from_identifier(FORK_IDENTIFIER); + for id in 2..=99 { + let identifier = format!("com.theerasakping.tpagentstudio.instance{id}"); + let other = RuntimeInstanceProfile::from_identifier(&identifier); + assert!(other.is_fork()); + assert_ne!(other.ide_server_port, primary.ide_server_port); + assert_ne!(other.cli_proxy_port, primary.cli_proxy_port); + assert!(other.default_orgii_home(Path::new("/home/test")).unwrap().to_string_lossy().contains("tp-agent-studio-instance")); + } + let webdriver = RuntimeInstanceProfile::from_identifier("com.theerasakping.tpagentstudio.e2e.instance2"); + assert_eq!(webdriver.ide_server_port, 23_848); + assert_eq!(webdriver.cli_proxy_port, 27_889); + } + #[test] fn dev_identifier_shares_primary_data_with_separate_service_defaults() { let dev = RuntimeInstanceProfile::from_identifier("org2ai.org2.dev"); diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index d8766d6b04..0c6c101c8c 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -1,8 +1,8 @@ { "$schema": "../node_modules/@tauri-apps/cli/config.schema.json", - "productName": "ORG2", + "productName": "TP Agent Studio", "version": "2.0.8", - "identifier": "org2ai.org2", + "identifier": "com.theerasakping.tpagentstudio", "build": { "frontendDist": "../build", "devUrl": "http://localhost:1998", @@ -14,7 +14,7 @@ "windows": [ { "label": "main", - "title": "ORG2", + "title": "TP Agent Studio", "width": 1200, "height": 800, "minWidth": 450, @@ -38,12 +38,19 @@ "csp": "default-src 'self'; script-src 'self' 'wasm-unsafe-eval'; style-src 'self' 'unsafe-inline' 'nonce-orgii-codemirror-style'; style-src-attr 'unsafe-inline'; img-src 'self' asset: http://asset.localhost https: data: blob:; font-src 'self' data:; connect-src 'self' asset: http://asset.localhost ipc: http://ipc.localhost ipc://localhost https: http://127.0.0.1:* http://localhost:* ws://127.0.0.1:* ws://localhost:* wss:; media-src 'self' asset: http://asset.localhost blob:; worker-src 'self' blob:; frame-src 'self' blob: asset: http://asset.localhost https: http://canvas-artifact.localhost canvas-artifact:; object-src 'none'; base-uri 'self'", "assetProtocol": { "enable": true, - "scope": ["$HOME/.orgii/**", "$HOME/.cursor/plugins/cache/**"] + "scope": [ + "$HOME/.tp-agent-studio/**", + "$HOME/.tp-agent-studio-dev/**", + "$HOME/.tp-agent-studio-instance*/**", + "$HOME/.cursor/plugins/cache/**" + ] } } }, "bundle": { - "externalBin": ["binaries/org2-pm"], + "externalBin": [ + "binaries/org2-pm" + ], "resources": { "resources/drag-icons/": "resources/drag-icons/" }, @@ -56,7 +63,7 @@ "icons/icon.icns", "icons/icon.ico" ], - "createUpdaterArtifacts": true + "createUpdaterArtifacts": false }, "plugins": { "fs": { @@ -64,14 +71,14 @@ }, "deep-link": { "desktop": { - "schemes": ["yorgai", "orgii"] + "schemes": [ + "tpagentstudio" + ] } }, "updater": { - "active": true, - "endpoints": [ - "https://github.com/org2AI/ORG2/releases/latest/download/latest.json" - ], + "active": false, + "endpoints": [], "dialog": false, "pubkey": "dW50cnVzdGVkIGNvbW1lbnQ6IG1pbmlzaWduIHB1YmxpYyBrZXk6IDExNEQ5NjIzMjA1ODI5OUMKUldTY0tWZ2dJNVpORWFiWXB3NW5UaWwyOHVxRXhzVmRVbXl2WktZNnhiN2tqcXlVY3gxTTlPV3EK" }, diff --git a/src-tauri/tauri.dev.conf.json b/src-tauri/tauri.dev.conf.json index 930d127a81..8a6c24e938 100644 --- a/src-tauri/tauri.dev.conf.json +++ b/src-tauri/tauri.dev.conf.json @@ -1,10 +1,12 @@ { - "productName": "ORG2 Dev", - "identifier": "org2ai.org2.dev", + "productName": "TP Agent Studio Dev", + "identifier": "com.theerasakping.tpagentstudio.dev", "plugins": { "deep-link": { "desktop": { - "schemes": ["yorgai-dev", "orgii-dev"] + "schemes": [ + "tpagentstudio-dev" + ] } }, "updater": { diff --git a/src-tauri/tauri.macos.conf.json b/src-tauri/tauri.macos.conf.json index b8286fc8cd..3189135e6e 100644 --- a/src-tauri/tauri.macos.conf.json +++ b/src-tauri/tauri.macos.conf.json @@ -5,9 +5,7 @@ }, "macOS": { "entitlements": "entitlements.plist", - "signingIdentity": "Developer ID Application: HOUYi HE (S4UG24G7HJ)", - "minimumSystemVersion": "10.15", - "providerShortName": "S4UG24G7HJ" + "minimumSystemVersion": "10.15" } } } diff --git a/src-tauri/tauri.windows.conf.json b/src-tauri/tauri.windows.conf.json index 9ea73fd4c7..a0fa4e0cc9 100644 --- a/src-tauri/tauri.windows.conf.json +++ b/src-tauri/tauri.windows.conf.json @@ -3,7 +3,7 @@ "windows": [ { "label": "main", - "title": "ORG2", + "title": "TP Agent Studio", "width": 1200, "height": 800, "minWidth": 450, diff --git a/src/config/runtimeInstance.test.ts b/src/config/runtimeInstance.test.ts index 1f2124dc8c..e5b034e643 100644 --- a/src/config/runtimeInstance.test.ts +++ b/src/config/runtimeInstance.test.ts @@ -3,6 +3,58 @@ import { describe, expect, it } from "vitest"; import { runtimeInstanceProfileForIdentifier } from "./runtimeInstance"; describe("runtimeInstanceProfileForIdentifier", () => { + it("isolates fork production ports and its registered auth scheme", () => { + expect( + runtimeInstanceProfileForIdentifier("com.theerasakping.tpagentstudio") + ).toEqual({ + instanceId: 1, + ideServerPort: 23_847, + cliProxyPort: 27_888, + authDeepLinkScheme: "tpagentstudio", + }); + }); + + it("keeps the fork dev identity separate from both production and upstream", () => { + expect( + runtimeInstanceProfileForIdentifier("com.theerasakping.tpagentstudio.dev") + ).toEqual({ + instanceId: 100, + ideServerPort: 23_946, + cliProxyPort: 27_987, + authDeepLinkScheme: "tpagentstudio-dev", + }); + }); + + it("matches the numbered and E2E fork profiles generated by the launcher", () => { + for (const identifier of [ + "com.theerasakping.tpagentstudio.instance2", + "com.theerasakping.tpagentstudio.e2e.instance2", + ]) { + expect(runtimeInstanceProfileForIdentifier(identifier)).toEqual({ + instanceId: 2, + ideServerPort: 23_848, + cliProxyPort: 27_889, + authDeepLinkScheme: "tpagentstudio-instance2", + }); + } + }); + + it("never routes unknown fork subidentifiers to upstream ports or orgii", () => { + for (const identifier of [ + "com.theerasakping.tpagentstudio.instance1", + "com.theerasakping.tpagentstudio.instance100", + "com.theerasakping.tpagentstudio.instance2.invalid", + "com.theerasakping.tpagentstudio.unknown", + ]) { + expect(runtimeInstanceProfileForIdentifier(identifier)).toEqual({ + instanceId: 1, + ideServerPort: 23_847, + cliProxyPort: 27_888, + authDeepLinkScheme: "tpagentstudio", + }); + } + }); + it("routes dev HTTP, WebSocket and login away from the bundled app", () => { expect(runtimeInstanceProfileForIdentifier("org2ai.org2.dev")).toEqual({ instanceId: 100, diff --git a/src/config/runtimeInstance.ts b/src/config/runtimeInstance.ts index db9fb2a0e8..441a4cb799 100644 --- a/src/config/runtimeInstance.ts +++ b/src/config/runtimeInstance.ts @@ -7,6 +7,9 @@ export interface RuntimeInstanceProfile { const PRIMARY_IDE_SERVER_PORT = 13_847; const PRIMARY_CLI_PROXY_PORT = 17_888; +const FORK_IDE_SERVER_PORT = 23_847; +const FORK_CLI_PROXY_PORT = 27_888; +const FORK_IDENTIFIER = "com.theerasakping.tpagentstudio"; // Dedicated dev slot, outside the numbered bundle range (2..99). const DEV_INSTANCE_ID = 100; @@ -18,7 +21,38 @@ const DEV_INSTANCE_ID = 100; export function runtimeInstanceProfileForIdentifier( identifier: string ): RuntimeInstanceProfile { - if (identifier.trim() === "org2ai.org2.dev") { + const normalized = identifier.trim(); + // Keep frontend runtime ports and OAuth callback schemes in parity with + // src-tauri/src/runtime_instance.rs and scripts/tauri/instance-profile.cjs. + // Unknown fork subidentifiers fail back to the FORK primary identity rather + // than accidentally selecting the upstream orgii:// / 13xxx endpoints. + if ( + normalized === FORK_IDENTIFIER || + normalized.startsWith(`${FORK_IDENTIFIER}.`) + ) { + const forkDev = normalized === `${FORK_IDENTIFIER}.dev`; + const suffix = normalized.slice(FORK_IDENTIFIER.length); + const match = /^(?:\.e2e)?\.instance(\d+)$/.exec(suffix); + const parsedId = match ? Number(match[1]) : 1; + const instanceId = forkDev + ? DEV_INSTANCE_ID + : Number.isInteger(parsedId) && parsedId >= 2 && parsedId <= 99 + ? parsedId + : 1; + const authDeepLinkScheme = forkDev + ? "tpagentstudio-dev" + : instanceId === 1 + ? "tpagentstudio" + : `tpagentstudio-instance${instanceId}`; + return { + instanceId, + ideServerPort: FORK_IDE_SERVER_PORT + instanceId - 1, + cliProxyPort: FORK_CLI_PROXY_PORT + instanceId - 1, + authDeepLinkScheme, + }; + } + + if (normalized === "org2ai.org2.dev") { return { instanceId: DEV_INSTANCE_ID, ideServerPort: PRIMARY_IDE_SERVER_PORT + DEV_INSTANCE_ID - 1, diff --git a/src/features/Org2Cloud/config.test.ts b/src/features/Org2Cloud/config.test.ts index 3bc820b3e1..d2f27731a5 100644 --- a/src/features/Org2Cloud/config.test.ts +++ b/src/features/Org2Cloud/config.test.ts @@ -86,6 +86,35 @@ describe("buildCloudAuthCallbackUrl", () => { ); }); + it("uses only fork-owned return schemes for fork production and dev builds", () => { + expect( + configureCloudAuthCallbackForIdentifier("com.theerasakping.tpagentstudio") + ).toBe("tpagentstudio://auth/callback"); + expect( + new URL(buildOrg2CloudLoginUrl()).searchParams.get("return_to") + ).toBe("tpagentstudio://auth/callback"); + + expect( + configureCloudAuthCallbackForIdentifier( + "com.theerasakping.tpagentstudio.dev" + ) + ).toBe("tpagentstudio-dev://auth/callback"); + expect( + new URL(buildOrg2CloudLoginUrl()).searchParams.get("return_to") + ).toBe("tpagentstudio-dev://auth/callback"); + }); + + it("derives the numbered and E2E fork callback scheme without using orgii", () => { + for (const identifier of [ + "com.theerasakping.tpagentstudio.instance2", + "com.theerasakping.tpagentstudio.e2e.instance2", + ]) { + expect(configureCloudAuthCallbackForIdentifier(identifier)).toBe( + "tpagentstudio-instance2://auth/callback" + ); + } + }); + it("derives an isolated callback from the runtime Tauri identifier", () => { expect( configureCloudAuthCallbackForIdentifier("org2ai.org2.instance2") diff --git a/src/hooks/platform/useDeepLinkHandler.test.ts b/src/hooks/platform/useDeepLinkHandler.test.ts index 05d87bd2a3..dd69187355 100644 --- a/src/hooks/platform/useDeepLinkHandler.test.ts +++ b/src/hooks/platform/useDeepLinkHandler.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { isUnclaimedCloudDeepLink, + parseDeepLink, reArmTrackedShareUrls, trackReArmableShareUrl, } from "./useDeepLinkHandler"; @@ -75,3 +76,41 @@ describe("malformed cloud deep-link containment", () => { expect(isUnclaimedCloudDeepLink("not a url")).toBe(false); }); }); + +describe("fork deep-link routing", () => { + it("normalizes the provisional fork schemes while keeping the in-app /orgii router base", () => { + expect( + parseDeepLink("tpagentstudio://marketplace/callback?code=abc") + ).toEqual({ + path: "/orgii/marketplace/callback", + search: "?code=abc", + }); + expect(parseDeepLink("tpagentstudio-dev://settings")).toEqual({ + path: "/orgii/settings", + search: "", + }); + expect( + parseDeepLink("tpagentstudio-instance2://project/open?id=3") + ).toEqual({ + path: "/orgii/project/open", + search: "?id=3", + }); + expect(parseDeepLink("orgii://marketplace/callback")).toEqual({ + path: "/orgii/marketplace/callback", + search: "", + }); + expect(parseDeepLink("unregistered://marketplace/callback")).toBeNull(); + }); + + it("does not let unhandled fork cloud links fall through to generic routing", () => { + expect(isUnclaimedCloudDeepLink("tpagentstudio://cloud/join?invite=")).toBe( + true + ); + expect( + isUnclaimedCloudDeepLink("tpagentstudio-dev://cloud/session?share=") + ).toBe(true); + expect( + isUnclaimedCloudDeepLink("tpagentstudio://marketplace/callback") + ).toBe(false); + }); +}); diff --git a/src/hooks/platform/useDeepLinkHandler.ts b/src/hooks/platform/useDeepLinkHandler.ts index 26872dd6e8..d54e0071e2 100644 --- a/src/hooks/platform/useDeepLinkHandler.ts +++ b/src/hooks/platform/useDeepLinkHandler.ts @@ -109,7 +109,7 @@ export function reArmTrackedShareUrls( * @param deepLinkUrl - The deep link URL (e.g., yorgai://marketplace/callback?code=xxx) * @returns Object with path and search, or null if invalid */ -function parseDeepLink( +export function parseDeepLink( deepLinkUrl: string ): { path: string; search: string } | null { try { @@ -128,7 +128,10 @@ function parseDeepLink( // earlier (see `parseCloudInviteDeepLink` / `parseCloudShareDeepLink`) // and never reach this generic conversion. - const withoutProtocol = deepLinkUrl.replace(/^(?:yorgai|orgii):\/\//, ""); + const supportedScheme = + /^(?:tpagentstudio(?:-dev|-instance\d+)?|yorgai|orgii):\/\//i; + if (!supportedScheme.test(deepLinkUrl)) return null; + const withoutProtocol = deepLinkUrl.replace(supportedScheme, ""); // Split path and query const [pathPart, ...queryParts] = withoutProtocol.split("?"); @@ -156,7 +159,8 @@ function parseDeepLink( // Only correct when called AFTER the dedicated cloud parsers had their turn. export function isUnclaimedCloudDeepLink(url: string): boolean { const trimmed = url.trim(); - if (!trimmed.toLowerCase().startsWith("orgii://")) return false; + if (!/^(?:orgii|tpagentstudio(?:-dev|-instance\d+)?):\/\//i.test(trimmed)) + return false; try { return ( new URL(trimmed).hostname.toLowerCase() === CLOUD_INVITE_DEEP_LINK_HOST