[Feature]: Reduce completion-observation latency in Task/Work waits

Start with openviking/service/task_tracker.py at the cited wait methods, then inspect Store implementations and the existing tracker, concurrency, Task/Work contract, and semantic lock ownership tests. The issue specifies a full Store notification design across FileStore and MySQL, race and cancellation behavior, and benchmark evidence; PR #7469 contains the implementation and is the sole PR to review. Done means the requested regression coverage, backend validation, latency and read-cost results, and semantic lock-path validation are included.

Issue Origin

Observed or reproduced in a real environment.

AI-assisted source review followed by a reported, executed component-level reproduction. The reproduction results below were supplied with this proposal and have not been independently rerun during its preparation. This is not a production incident report.

Problem Statement

TaskTracker.wait(), wait_for_work(), and wait_for_task_work() use exponential polling, starting at 50 ms and capped at 10 seconds.

Once a waiter reaches a long sleep interval, a state change satisfying its completion condition does not trigger an early recheck. The waiter can therefore remain asleep for nearly another 10 seconds after the required work has finished.

A component-level reproduction on commit 6c00989c7fdb97e8b5378e6cc8fc1c083b37d9b4 reported:

Event	Elapsed time
Terminal state confirmed	13.003 s
wait() returned	22.777 s
wait_for_work() returned	22.777 s
wait_for_task_work() returned	22.777 s
Additional observation delay	Approximately 9.775 s

The polling implementation is also present in the source reviewed at commit 226892efcabe99d08b7b7ed3f7c72c3379377b96. The timing results above belong to the earlier reproduction commit.

Sources: Task wait, Work waits.

Use Case

This affects existing execution paths:

• Resource ingestion with wait=true waits through TaskTracker.wait().
• A semantic message with a lock handoff waits for its Embedding Work before closing its lock scope.
• AddResource and Reindex processing wait for downstream Work before continuing their completion logic.

Delayed observation can therefore postpone request completion and internal processing, including lock-scope closure. The reproduction measures waiter latency; it does not establish a production throughput improvement or quantify lock contention.

Source: Semantic Embedding wait and lock closure.

Evidence and Reproduction

Reported environment:

• OpenViking commit: 6c00989c7fdb97e8b5378e6cc8fc1c083b37d9b4
• Python: 3.12.14
• OS: Linux

The reproduction uses the unmodified TaskTracker and PersistentTaskStore, including their real state transitions. Only AGFS I/O is replaced by an in-memory fake. Wait methods, polling constants, clocks, and asyncio.sleep are unchanged.

It has not been benchmarked through a live OpenViking server, native AGFS, or a real MySQL deployment.

Save the following as test_scripts/repro_task_wait_latency.py. From the reproduction checkout with its dependencies installed, run:

PYTHONPATH=. python test_scripts/repro_task_wait_latency.py

import asyncio
import time

from openviking.service.task_store import PersistentTaskStore
from openviking.service.task_tracker import TaskStatus, TaskTracker

OWNER = {"account_id": "account", "user_id": "user"}


class FakeAgfs:
    def __init__(self):
        self.files = {}

    def mkdir(self, path, **kwargs):
        return {}

    def read(self, path, **kwargs):
        if path not in self.files:
            raise FileNotFoundError(path)
        return self.files[path]

    def write(self, path, data, **kwargs):
        self.files[path] = data
        return "OK"


async def main():
    tracker = TaskTracker(PersistentTaskStore(FakeAgfs()))
    await tracker.create("add_resource", task_id="task", **OWNER)
    await tracker.start("task", **OWNER)
    await tracker.register_work("task", "embedding", "Embedding", **OWNER)
    await tracker.complete("task", {}, **OWNER)
    assert (await tracker.get("task", **OWNER)).status == TaskStatus.RUNNING

    started = time.monotonic()
    committed = None

    async def finish():
        nonlocal committed
        await asyncio.sleep(13.0)
        await tracker.mark_work_done("task", "embedding", **OWNER)
        assert (await tracker.get("task", **OWNER)).status == TaskStatus.COMPLETED
        committed = time.monotonic()
        print(f"terminal state confirmed: {committed - started:.3f}s")

    async def observe(name, waiter):
        await waiter
        return name, time.monotonic()

    results = await asyncio.gather(
        observe("wait", tracker.wait("task", **OWNER)),
        observe(
            "wait_for_work",
            tracker.wait_for_work("task", ["embedding"], **OWNER),
        ),
        observe(
            "wait_for_task_work",
            tracker.wait_for_task_work("task", **OWNER),
        ),
        finish(),
    )
    for name, returned in results[:-1]:
        print(
            f"{name}: returned={returned - started:.3f}s, "
            f"lag={returned - committed:.3f}s"
        )


asyncio.run(main())

Proposed Solution

Allow existing wait methods to recheck their completion conditions promptly after relevant authoritative state changes, while retaining bounded periodic reconciliation.

The proposed initial boundary is:

• Callers sharing one Store instance receive local wakeups, including across threads and event loops.
• Notifications are hints to recheck state. Existing Store queries remain authoritative for completion.
• Independent Store instances and other processes remain observable through periodic rechecks.
• Notification delivery does not reset timeout budgets or postpone reconciliation indefinitely.

Notification publication belongs to the Store:

• FileStore publishes after durable Task updates become visible in its cache, or after successful process-local Work updates.
• MySQL publishes after successful transaction commit.
• A confirmed successful update must still publish when caller cancellation is propagated afterward.
• Rollbacks, rejected operations, and unchanged duplicate operations do not publish state changes.
• Metadata-only updates that cannot affect these waits do not independently trigger notifications.

The proposal does not add a persisted revision, schema migration, distributed broker, SSE endpoint, or SDK/CLI waiting protocol. Those would require separate requirements.

Behavior and Failure Requirements

1. Preserve the three predicates:

   • wait() waits for a terminal Task and returns its record, including failure or cancellation.
   • wait_for_work(ids) waits only for the specified Work IDs and preserves missing-record errors.
   • wait_for_task_work() queries the current open Work set, including recovered or newly registered Work and exclude_work_id.

2. Register before reading authoritative state so completion before registration, during the read, or immediately before suspension cannot leave the waiter sleeping until reconciliation.

3. Preserve completion precedence over timeout or stop checks, empty-ID behavior, shared timeout budgets, cancellation, and ownership boundaries.

4. Clean up registrations after completion, errors, timeout, and cancellation. Cancelling one waiter must not affect another.

5. Handle closed waiter loops without changing a committed writer’s result or preventing delivery to other waiters.

6. Preserve progress when notifications are missed, updates come from an independent MySQL Store or process, or stop_waiting changes without a Task/Work mutation.

7. Preserve existing behavior when records are withdrawn, deleted, or expired. Notifications must not manufacture a successful result or hide Store errors.

Validation

• Add focused regression coverage to the existing tracker, concurrency, Task/Work contract, and semantic lock-ownership tests.
• Use synchronization barriers for registration races and cross-thread tests.
• Exercise both FileStore and real MySQL. A skipped MySQL test does not establish MySQL correctness.
• Demonstrate that local completion interrupts a long reconciliation interval.
• Compare the baseline and implementation under the same environment, reporting completion-to-return p50/p95/max for all three waits with 1, 10, and 100 waiters.
• Report logical state-read calls separately from AGFS operations and MySQL SQL executions.
• Include idle waiting, unrelated changes, duplicate submissions, and frequent Work completions. Do not assume notification fan-out always reduces reads.
• Validate the existing semantic path: related Embedding Work must finish before lock closure, and observation should no longer wait for the remaining local polling interval.

The previously proposed benchmark target of local p95 ≤100 ms and maximum <1 second is a discussion target, not an established service guarantee.

Proposed Delivery

Deliver the complete change in a single PR, #7469:

• Add Store-owned notifications and use them in TaskTracker.wait() for both backends, including Task finalization caused by Work completion.
• Use the same mechanism in wait_for_work() and wait_for_task_work(), publish Work-only changes, and include semantic lock-path validation.
• Include the associated contract tests, reproduction script, performance/read-cost evidence and disclosed check limitations in the same PR.

PR #7469 already contains both implementation commits and is the sole PR to review for this issue. It supersedes #7467, which is being closed without merging; there is no dependency on merging #7467 first. The earlier two-PR delivery plan is replaced by this single-PR plan. PR #7469 remains Draft pending the outstanding checks and review follow-up; this delivery update does not claim maintainer approval.

Alternatives Considered

• Reducing the polling cap improves observation latency but increases steady-state polling.
• Pure notification waiting needs a recovery path for independent writers, missed signals, and worker-stop changes.
• Distributed notifications and public subscriptions expand the API and deployment scope beyond this proposal.

Feature Area

Core (Client/Engine)

The implementation is available in Draft PR #7469. Feedback on the Store ownership boundary and the complete Task/Work scope is welcome; no maintainer approval is claimed.