Complete async query-job waiting and result handoff

.github/workflows/windows-private-storage.yml

@@ -0,0 +1,16 @@
+name: Windows private storage
+on:
+  push:
+  pull_request:
+  workflow_dispatch:
+jobs:
+  windows-private-storage:
+    runs-on: windows-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v5
+      - uses: astral-sh/setup-uv@v6
+        with:
+          version: "0.12.13"
+      - run: uv sync --locked --python 3.13
+      - run: uv run python -m unittest discover -s tests -p test_private_storage_windows.py -v

README.md

@@ -29,9 +29,15 @@ job-routing instructions; the skill folders can also be installed separately.
 ## Long queries, CLI and progress
 
 Use `omop-submit_query_job` for an export or a query that could exceed the
-interactive timeout. It returns a job ID immediately. Poll
-`omop-query_job_status` at its recommended interval; use
-`omop-cancel_query_job` to stop. Only `completed` means the file is complete.
+interactive timeout. Call `omop-query_job_capacity` first; submission returns a
+job ID immediately and atomically checks the two-job limit. Use
+`omop-wait_for_query_job` for native waits of up to 30 seconds, or
+`omop-query_job_status` for an immediate snapshot. Use `omop-cancel_query_job`
+to stop. Only `completed` means the file is complete. Read bounded UTF-8 chunks
+with `omop-read_query_job_result` using the job ID and `next_offset`; do not pass
+`result_path` to generic filesystem tools. The secure reader requires POSIX;
+unsupported platforms retain the local CLI file. See [job boundaries and the
+credential-free app fixture](docs/QUERY_JOBS.md#mcp-waiting-capacity-and-result-handoff).
 Status includes rows, bytes, elapsed time, phase and an advisory ETA when known.
 Set `mode="explain"` to obtain a plan without running the query.
 
@@ -48,7 +54,8 @@ uv run ucsfomopagent watch JOB_ID
 
 No-argument `uv run ucsfomopagent` continues to start the MCP server. `status`, `watch`,
 `list`, `cancel` and `purge` need no database credentials. Results remain in the
-local private job directory and are not sent to chat. Database/server limits can
+local private job directory. The bounded result reader can return requested chunks
+to chat; keep controlled data local and share only appropriate summaries. Database/server limits can
 still fail a query; jobs report those failures instead of silently truncating.
 
 ### Large fact-table studies

docs/QUERY_JOBS.md

@@ -13,8 +13,9 @@ Keep each connector independently installable and releasable. Database adapters
 live in query_backend.py, orchestration in jobs.py/job_worker.py, and command
 parsing in query_cli.py. Existing domain tools remain alongside them. This avoids
 a new separately published dependency or any requirement to clone sibling repos.
-The three small job-runtime modules intentionally have identical source; when
-changing the protocol, run the same contract tests in all three repositories.
+OMOP and CDW keep jobs.py and job_worker.py identical. Their query adapters differ
+only in connector defaults. Run the same contract tests in both repositories when
+changing the protocol; the SPOKE connector has an independently released runtime.
 A future shared package is appropriate once its API and release ownership settle.
 
 `extensions/ucsfomopagent.brxt` is the current committed installable artifact. Build it
@@ -46,8 +47,15 @@ malicious code running under the same OS account.
 ## Storage and progress
 
 Jobs live in `~/.local/share/ucsf-agents/ucsfomopagent/` (override the parent using
-UCSF_AGENT_JOB_ROOT). Directories are private (0700) and DB/results are 0600 on
-POSIX. On Windows use the account's private profile directory and its ACLs.
+UCSF_AGENT_JOB_ROOT). The storage backend checks current-account ownership,
+rejects symlinks, nonregular entries and hard-linked private files, and repairs
+existing owned stores before reading them. On macOS/Linux it uses retained
+no-follow descriptors, modes 0700/0600 and removal of inherited ACL grants.
+New result files are protected before any result bytes are written. Windows uses
+a protected current-user DACL through retained no-reparse handles. Unsupported
+or unsafe storage returns an actionable `private_storage_unavailable` error from
+the job tools; synchronous connector tools remain available. Normal job use
+requires no additional prompts or changes to the six-tool interface.
 SQLite serializes admission (two active jobs per connector). Results stream in
 1000-row batches to a `.partial` file and are renamed only after success and
 fsync. The default disk budget is 10 GB; exceeding it fails visibly rather than
@@ -56,6 +64,30 @@ and JSON-encodes graph/list values. Use JSONL when empty string and null must be
 distinguished. SQL projections require unique column aliases. Exported text may
 contain spreadsheet formula strings: treat it as data when importing CSV.
 
+These permissions protect against other ordinary local accounts where filesystem
+ACLs are enforced; they do not revoke file handles another process opened before
+an old store was repaired. They do not isolate arbitrary code running under the
+same OS account, administrators or root. Choose storage beneath a private or
+administrator-controlled parent directory. SQLite and process-lease/final-rename
+operations still use pathnames after identity checks; this update does not claim
+to eliminate every pathname replacement race in the complete job runtime.
+
+On Windows, automatic repair first checks every managed directory for mutation
+rights granted to another ordinary account. A writable known child remains
+accessible to that account even when its parent is private, so a protected parent
+alone does not make legacy migration safe. Such stores return
+`private_storage_requires_quiescent_repair` before changing existing ACLs. Unknown
+ACL formats receive the same conservative refusal. Already-private stores and
+legacy read-only inheritance continue through normal automatic repair.
+
+For this refusal, stop the connector and other writers and ask an administrator
+to repair only the owned ordinary entries without following links, verify that
+other ordinary accounts cannot mutate any managed directory, and retry. Another
+option is a fresh private `UCSF_AGENT_JOB_ROOT`, retaining the old store untouched
+for trusted recovery. Moving an old directory alone does not repair its existing
+ACLs. The connector does not delete old data or silently migrate an actively
+shared store. Ordinary job use needs no additional approval prompts.
+
 Status records queued/running/terminal states and separate connecting/planning/
 executing/streaming phases,
 rows written, bytes, heartbeat, elapsed time and advisory ETA. A heartbeat proves
@@ -79,6 +111,78 @@ uv run ucsfomopagent cancel JOB_ID
 uv run ucsfomopagent purge JOB_ID
 ```
 
+## MCP waiting, capacity and result handoff
+
+Six namespace-prefixed job tools are available. `submit_query_job` starts work;
+`query_job_status` is an immediate lookup; `cancel_query_job` requests cancellation.
+`query_job_capacity` reports `active_jobs`, `max_active_jobs` (2), and
+`available_slots`. Admission is checked atomically, so a later submission can
+still return `capacity_exceeded` with current capacity. Waiting jobs hold their
+slot until completion, failure or confirmed cancellation.
+
+`wait_for_query_job(job_id, timeout_seconds=30)` waits natively for 0–30 seconds.
+It returns a current status plus `wait_timed_out`; true means the wait budget ended
+while the job remains active. Cancelling the MCP wait does not cancel the job.
+Use another bounded wait or respect `recommended_poll_seconds` for immediate
+status polls. Status lookups run off the MCP event loop with a short database
+lock budget. If the wait expires while storage is busy after an active state
+was observed, it returns that last state with `wait_timed_out=true`,
+`wait_status_stale=true` and an explicit warning. If no state could be read,
+it returns a transient `status_lookup_busy` lookup error without claiming an
+operation outcome. A busy lookup never marks the job failed. No shell sleep,
+fabricated progress or automatic resubmission is needed. Unknown ETA is normal
+during connection, planning and execution.
+
+`read_query_job_result(job_id, offset=0, max_bytes=8192)` reads only a completed
+job's managed CSV/JSONL file. It accepts no path. It returns `text`, byte
+`offset`, `bytes_read`, `total_bytes`, `eof` and `next_offset`. Limits are 4–65536
+bytes per call; follow `next_offset` until `eof=true`. Chunk boundaries preserve
+UTF-8 characters but can split rows or quoted CSV records. Assemble the desired
+bounded content before parsing it. A mid-character offset is rejected.
+
+The reader ignores saved `result_path` values and opens a fixed result name
+relative to the managed job directory, using pinned-root identity, descriptor-
+relative opens and no-follow flags. It rejects directory/file symlinks, hardlinks,
+nonregular files and changed result sizes. POSIX secure-open primitives are
+required. On unsupported platforms (including Windows) it fails closed with
+`secure_reader_unavailable`; the existing local CLI result remains available.
+This does not grant arbitrary filesystem access or isolate hostile code running
+under the same OS account. Keep controlled data local and use bounded summaries
+rather than placing complete patient-level exports in chat.
+
+Failed jobs retain `error_details` with stable `code`, `kind`, sanitized `message`,
+original failure `phase`, and boolean `retryable`. Driver messages and query
+literals are not exposed. Retryability is advisory; resubmission remains explicit.
+New submissions, status/wait responses and cancellation responses also carry:
+
+```json
+{"_meta":{"io.biorouter/operation":{"version":1,"id":"ucsfomopagent:JOB_ID","status":"failed","error":{"code":"query_rejected","kind":"invalid_args","message":"The database rejected the query. Check schema, projections and parameters.","phase":"executing","retryable":false}}}}
+```
+
+Allowed operation states are `queued`, `running`, `completed`, `failed` and
+`cancelled`; job `timed_out` and `interrupted` map to operation `failed`. A status
+lookup that finds a failed job still has MCP `isError=false`. Unknown-job,
+admission and result-reader failures have `isError=true` and structured safe
+errors, without changing the stored job outcome. Older failed records without
+structured details receive a generic safe operation error.
+
+## Credential-free app smoke fixture
+
+From a source checkout, create a fresh empty absolute directory and run:
+
+```bash
+uv run python scripts/synthetic_job_server.py --root /absolute/fresh-empty-dir --delay-seconds 2
+```
+
+This stdio MCP server registers the actual six job tools with an isolated store.
+It accepts only `SELECT 1 AS example` (three synthetic multilingual rows),
+`SELECT 0 AS simulate_failure` (stable safe failure), and
+`SELECT 2 AS simulate_slow` (bounded delayed success). No real database connection,
+credential lookup or installed job store is used. An existing nonempty directory
+is refused. Use `omop-submit_query_job`, wait/status, and the result reader to
+exercise real MCP metadata and chunked handoff in a development app. Use separate
+fresh roots for concurrent fixtures.
+
 ## Read-only and failure boundaries
 
 Use database accounts with SELECT/read-only privileges. The shared query guard
@@ -130,3 +234,17 @@ Security updates remove the former Intel-macOS cryptography<49 constraint.
 Patched cryptography 50+ is required. If a compatible Intel macOS wheel is not
 available, install the upstream Rust/OpenSSL build prerequisites or use a
 supported Python/platform combination; do not restore a vulnerable version cap.
+
+## Submission and CLI follow-up semantics
+
+Capacity responses and rejection context include `advisory: true`: another client
+may consume an available slot before submission. Atomic admission remains the
+authority. Once a job ID has been allocated, a local worker-start or pipe failure
+returns that durable ID with failed operation metadata; it does not hide the job
+behind a generic submission error. No active slot remains allocated to that failure.
+
+CLI `watch JOB_ID` uses the same native bounded wait as MCP instead of a separate
+sleep/poll loop. The wait remains at most 30 seconds per call and can be repeated
+until terminal. `status JOB_ID` remains an immediate lookup. Terminal payload
+`phase` is the terminal state, while `error_details.phase` identifies where the
+failure actually occurred (for example `executing` or `worker_start`).

extensions/SHA256SUMS

@@ -1 +1 @@
-62f53afd87a1378a65e89b04c0de453123c450e499543bbe2a25a81345438e6a  ucsfomopagent.brxt
+0eab7151e1005693205dcd76bded202a52dbeca33787e4c3ba99d3c790ad22cf  ucsfomopagent.brxt

scripts/synthetic_job_server.py

@@ -0,0 +1,149 @@
+"""Credential-free MCP job fixture; only the three documented queries are accepted."""
+from __future__ import annotations
+
+import argparse
+import importlib
+import json
+import math
+from pathlib import Path
+import sys
+import threading
+import time
+
+REPOSITORY = Path(__file__).resolve().parents[1]
+PACKAGE = json.loads((REPOSITORY / 'manifest.json').read_text())['name']
+PREFIXES = {'ucsfomopagent': 'omop-', 'cdwagent': 'CDW-'}
+if PACKAGE not in PREFIXES:
+    raise RuntimeError('This fixture supports only the OMOP and CDW connector packages.')
+PREFIX = PREFIXES[PACKAGE]
+sys.path.insert(0, str(REPOSITORY / 'src'))
+
+jobs = importlib.import_module(PACKAGE + '.jobs')
+worker = importlib.import_module(PACKAGE + '.job_worker')
+
+EXAMPLE_QUERY = 'SELECT 1 AS example'
+FAILURE_QUERY = 'SELECT 0 AS simulate_failure'
+SLOW_QUERY = 'SELECT 2 AS simulate_slow'
+ALLOWED_QUERIES = frozenset((EXAMPLE_QUERY, FAILURE_QUERY, SLOW_QUERY))
+EXAMPLE_ROWS = ((1, 'café'), (2, '漢字 😀'), (3, 'a, "quote"\nsecond line'))
+SYNTHETIC_CREDENTIALS = dict(
+    server='synthetic.invalid', database='synthetic',
+    username='synthetic-only', password='synthetic-only',
+)
+
+
+def bounded_delay(value):
+    try:
+        delay = float(value)
+    except (TypeError, ValueError):
+        raise ValueError('delay_seconds must be a number from 0 to 30.') from None
+    if isinstance(value, bool) or not math.isfinite(delay) or not 0 <= delay <= 30:
+        raise ValueError('delay_seconds must be a number from 0 to 30.')
+    return delay
+
+
+class SyntheticJobStore(jobs.JobStore):
+    def __init__(self, root, delay_seconds=2):
+        self.delay_seconds = bounded_delay(delay_seconds)
+        root = Path(root)
+        if not root.is_absolute() or root.is_symlink():
+            raise ValueError('Supply an absolute, fresh temporary directory as root.')
+        if root.exists() and (not root.is_dir() or any(root.iterdir())):
+            raise ValueError('The synthetic job root must be empty; use a fresh temporary directory.')
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        super().__init__(PACKAGE, root)
+        self._threads = []
+        self._threads_lock = threading.Lock()
+        self._slow_release = threading.Event()
+        self._closed = False
+
+    def submit(self, query, parameters, credentials, **options):
+        if (not isinstance(query, str) or query.strip() not in ALLOWED_QUERIES
+                or parameters not in (None, {}) or options.get('mode', 'query') != 'query'):
+            raise jobs.JobError(
+                'synthetic_query_rejected', 'invalid_args',
+                'This fixture accepts only its three documented queries without parameters.',
+                'validation',
+            )
+        if self._closed:
+            raise jobs.JobError('synthetic_server_closed', 'internal',
+                                'The synthetic job server is closed.', 'admission')
+        return super().submit(query.strip(), {}, dict(SYNTHETIC_CREDENTIALS), **options)
+
+    def _stream(self, request, credentials, progress):
+        if request['query'] == FAILURE_QUERY:
+            raise jobs.JobError('synthetic_failure', 'tool_failure',
+                                'Synthetic job failed as requested.', 'executing')
+        if request['query'] == SLOW_QUERY:
+            progress['phase'] = 'executing'
+            self._slow_release.wait(self.delay_seconds)
+        yield ['example', 'label'], EXAMPLE_ROWS
+
+    def _start_worker(self, job_id, request, credentials):
+        def run():
+            try:
+                with self.lease(job_id).acquire(timeout=0):
+                    worker.execute(self, job_id, dict(request=request, credentials={}), self._stream)
+            except Exception:
+                data = self.status(job_id)
+                jobs.record_failure(data, dict(
+                    code='synthetic_worker_failed', kind='internal',
+                    message='The synthetic worker could not complete.',
+                    phase='synthetic_worker', retryable=False,
+                ))
+                self.update(data)
+
+        thread = threading.Thread(target=run, name='synthetic-query-job', daemon=True)
+        with self._threads_lock:
+            self._threads.append(thread)
+            thread.start()
+
+    def release_slow_jobs(self):
+        self._slow_release.set()
+
+    def close(self):
+        self._closed = True
+        self.release_slow_jobs()
+        deadline = time.monotonic() + 5
+        with self._threads_lock:
+            threads = list(self._threads)
+        for thread in threads:
+            thread.join(max(0, deadline - time.monotonic()))
+        return not any(thread.is_alive() for thread in threads)
+
+
+def create_server(root, delay_seconds=2):
+    from fastmcp import FastMCP
+    store = SyntheticJobStore(root, delay_seconds)
+    server = FastMCP(
+        PACKAGE + ' synthetic jobs',
+        instructions=(
+            'This is a synthetic fixture with no database access or real records. '
+            'Allowed queries are exactly SELECT 1 AS example, SELECT 0 AS simulate_failure, '
+            'and SELECT 2 AS simulate_slow. Use query_job_capacity, submit_query_job, '
+            'wait_for_query_job and read_query_job_result to exercise the job contract.'
+        ),
+    )
+    jobs.register_job_tools(server, PACKAGE, dict(SYNTHETIC_CREDENTIALS), PREFIX, store=store)
+    return server, store
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--root', type=Path, required=True,
+                        help='Absolute path to a fresh, empty temporary directory.')
+    parser.add_argument('--delay-seconds', type=float, default=2,
+                        help='Synthetic slow-job delay, from 0 to 30 seconds (default: 2).')
+    arguments = parser.parse_args(argv)
+    try:
+        server, store = create_server(arguments.root, arguments.delay_seconds)
+    except ValueError as error:
+        parser.error(str(error))
+    try:
+        server.run(transport='stdio')
+    finally:
+        store.close()
+
+
+if __name__ == '__main__':
+    main()