Repository navigation
Feat/executor cleanup cursor - #1014
Conversation
…eanup Introduce cursor-based executor cleanup with distributed locking, Redis caching, and progress tracking. Refactor executor job cleanup logic and add comprehensive tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
📝 WalkthroughWalkthroughConverts executor cleanup from synchronous to asynchronous: adds async DB sessions, async Redis/distributed-lock APIs, cursor-based batched cleanup scans, expanded filtering and recovery logic, and updates background worker, archive handling, and tests to async flows. Changes
Sequence DiagramsequenceDiagram
participant API as FastAPI Endpoint
participant Lock as DistributedLock
participant Redis as Redis
participant Job as JobService
participant DB as AsyncSession
participant Exec as ExecutorService
API->>Lock: acquire_watchdog_context_async(lock_key)
Lock->>Redis: SET ... NX EX (acquire)
Redis-->>Lock: acquired / contended
alt Lock acquired
Lock->>Job: cleanup_stale_executors(db)
Job->>Redis: get_cursor()
Redis-->>Job: cursor or init
loop batched scanning
Job->>DB: query Subtask batch by cursor
DB-->>Job: candidate subtasks
Job->>Job: filter candidates (rules/stats)
Job->>DB: load TaskResource for task_ids
DB-->>Job: active tasks
Job->>Exec: delete_executor_task_async(...)
Exec-->>Job: success / error
Job->>Redis: advance_cursor(last_scanned_subtask_id)
Redis-->>Job: persisted
end
Lock->>Lock: extend_async() periodically (watchdog)
Lock->>Redis: EXPIRE (extend)
end
Lock->>Redis: DEL (release)
Lock-->>API: complete
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 450cb37f92
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if not task: | ||
| continue | ||
|
|
||
| subtasks = await self._get_cleanup_subtasks_for_task(db, task_id) |
There was a problem hiding this comment.
_cleanup_executor_groups reloads all undeleted subtasks for each task via _get_cleanup_subtasks_for_task instead of reusing valid_candidates. This means when a task has one stale executor subtask and another recently updated one, the recent executor is deleted too, bypassing the code_task_recent / task_updated_recent retention checks and causing premature executor teardown.
Useful? React with 👍 / 👎.
| Subtask.created_at > lookback_start, | ||
| Subtask.created_at <= cutoff, |
There was a problem hiding this comment.
The lookback query is bounded by created_at, but actual stale eligibility is based on updated_at <= chat_cutoff in _filter_scanned_subtasks. A long-lived subtask created before the lookback window but updated recently can be skipped in primary scan, have the cursor move past it, and then never be revisited once it becomes stale, leaving its executor undeleted indefinitely.
Useful? React with 👍 / 👎.
|
|
||
|
|
||
| def stop_background_jobs(app): | ||
| async def stop_background_jobs(app): |
There was a problem hiding this comment.
This function was changed to async, but the shutdown path in backend/app/main.py still invokes stop_background_jobs(app) without await. In practice that returns an un-awaited coroutine, so the body here does not run during shutdown and background workers are not actually stopped gracefully.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (3)
backend/app/core/distributed_lock.py (1)
337-375: Watchdog task: minor — cancel-then-await pattern is correct.Cancelling the watchdog and awaiting it under
CancelledErrorsuppression is the right shutdown dance. One small suggestion: ifacquired is Truebut the watchdog never started (e.g.extend_interval_seconds >= expire_seconds), we silently lose the extension guarantee. Consider logging a warning in that branch so misconfigured intervals are visible.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/core/distributed_lock.py` around lines 337 - 375, In acquire_watchdog_context_async: when acquired is True but watchdog_task is None (meaning extend_interval_seconds is <=0 or >= expire_seconds or async_redis_client is None) add a warning log to make the misconfiguration visible; locate the check that creates watchdog_task (uses acquired, self.async_redis_client, extend_interval_seconds, expire_seconds) and, after that block, if acquired and watchdog_task is None, call the module logger or self.logger to emit a clear warning mentioning lock_name, expire_seconds, and extend_interval_seconds so users know the extension watchdog was not started.backend/app/services/workspace_archive/archive_service.py (1)
378-403: Avoid reaching intoarchive_storage_service._bucket.
_bucketis a private attribute of another module; consuming it here couples the recovery logic toArchiveStorageService’s internals and will silently break if that attribute is renamed/removed. Expose a small helper onarchive_storage_service(e.g.stat_archive(storage_key) -> Optional[ObjectStat]orget_bucket()) and call it here.♻️ Sketch
- stat = archive_storage_service.client.stat_object( - archive_storage_service._bucket, storage_key - ) + stat = archive_storage_service.stat_archive(storage_key) + if stat is None: + return None…with
stat_archiveimplemented instorage.pyusing the existing bucket/client.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/workspace_archive/archive_service.py` around lines 378 - 403, The _try_recover_archive method is directly accessing archive_storage_service._bucket which couples this code to that module's internals; add a public helper on archive_storage_service (e.g. stat_archive(storage_key) that uses the existing client and bucket internally and returns the object stat or None) and replace the direct call to archive_storage_service.client.stat_object(archive_storage_service._bucket, storage_key) with a call to archive_storage_service.stat_archive(storage_key), then use the returned stat.size as before and keep the same logging/exception handling in _try_recover_archive.backend/tests/services/test_preserve_executor.py (1)
1-1401: Split this oversized test module.This file is now 1401 lines. Consider splitting stale cleanup, manual cleanup API, preserve-label API, and label consistency tests into separate modules. As per coding guidelines, “File size limit: if a file exceeds 1000 lines, split it into multiple sub-modules”.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/tests/services/test_preserve_executor.py` around lines 1 - 1401, The test module is too large (>1000 lines); split it into smaller modules by logical area: create separate test files for stale cleanup scans (containing CleanupExecutorTestHelpers and class TestCleanupStaleExecutorsWithPreserveFlag), manual/task-scoped API cleanup (TestCleanupTaskExecutorAPI), preserve-label API tests (TestSetPreserveExecutorAPI), and label consistency tests (TestPreserveExecutorLabelConsistency). Move related fixtures and helper methods (e.g., _create_mock_task_resource, _create_mock_subtask, _patch_cleanup_scan) into the appropriate new modules or a shared conftest/helper module so each new file imports only what it needs, update imports/patch targets (JobService, TaskKindsService, TaskResource, Subtask, settings, executor_kinds_service) and run the test suite to ensure no import/fixture breakage.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/core/distributed_lock.py`:
- Around line 258-278: The async Redis client property async_redis_client
currently caches an AsyncRedis instance via AsyncRedis.from_url without
verifying connectivity or guarding against event-loop binding; update the
async_redis_client getter to create the client, perform await client.ping() to
fail fast if Redis is unreachable (mirroring the sync redis_client behavior),
and then only cache the client if ping succeeds; additionally, to avoid
cross-event-loop RuntimeError, either stop global caching and create a fresh
AsyncRedis via AsyncRedis.from_url per call in acquire_async/extend_async
(matching the per-call pattern in app/core/cache.py) or implement a cache keyed
by the current loop id (use id(asyncio.get_running_loop())) so the stored client
is tied to the calling event loop.
In `@backend/app/db/session.py`:
- Around line 8-9: The URL-to-async conversion currently concatenates strings
and produces invalid dialects (e.g., mysql+mysqldb -> mysql+asyncmy+mysqldb);
instead parse the DB URL with sqlalchemy.engine.make_url (or make_url) and
rebuild the drivername properly: extract the dialect and existing driver from
the parsed URL, replace the sync driver with the intended async driver (keeping
the dialect prefix), assign the new drivername back to the parsed URL, and pass
that URL to create_async_engine/AsyncEngine/AsyncSession creation code (the code
around AsyncEngine/create_async_engine in session.py). Update the conversion
logic to use make_url so SQLAlchemy receives a valid URL rather than a simple
string concat.
In `@backend/app/services/adapters/executor_job.py`:
- Around line 285-303: The filter in _filter_scanned_subtasks currently allows
subtasks with an empty or None executor_namespace even though scheduled cleanup
requires both executor_name and executor_namespace; update the list
comprehension to also require a truthy subtask.executor_namespace (and keep the
other checks: status in {PENDING, COMPLETED, FAILED, CANCELLED},
isinstance(updated_at, datetime) and updated_at <= chat_cutoff, truthy
executor_name, not _is_device_executor_name(executor_name), and not
executor_deleted_at) so no subtasks with missing namespaces are forwarded to
delete_executor_task_async; apply the same executor_namespace truthy check to
the analogous filter block referenced around lines 527-539.
- Around line 127-151: The cursor is advanced too early: currently
executor_cleanup_cursor_service.advance_cursor(...) is called right after
computing last_id from scanned_subtasks but before calling
_process_cleanup_batch(...); move the advance_cursor call to after
_process_cleanup_batch returns successfully (after
batch_valid_count/batch_deleted_count/batch_filter_stats are obtained) so that
only fully processed subtasks mark the cursor; keep using last_id computed from
scanned_subtasks and only call advance_cursor when _process_cleanup_batch
completes without raising.
- Around line 263-279: Replace direct boolean equality checks against False with
SQLAlchemy's is_ comparison: change occurrences of Subtask.executor_deleted_at
== False to Subtask.executor_deleted_at.is_(False) in the query used with
db.execute/select(Subtask) (the block assigning result). Also update the same
pattern at the other two occurrences referenced (around the logic near lines
shown — the other similar Subtask queries at the later usages) so all checks use
.is_(False) instead of == False.
In `@backend/app/services/executor_cleanup_cursor_service.py`:
- Around line 40-64: The threshold uses naive datetime.now() in
_build_default_cursor which can be timezone-mismatched with Subtask.created_at
(populated by func.now()); replace the naive timestamp generation for
recent_threshold with a timezone-consistent value (e.g., use timezone-aware UTC
now such as datetime.now(timezone.utc) or datetime.utcnow() converted to aware)
so comparisons against Subtask.created_at are aligned, and apply the same change
wherever similar cutoff logic exists (refer to recent_threshold,
Subtask.created_at, and ExecutorCleanupCursor in this service and the related
executor_job.py).
In `@backend/app/services/jobs.py`:
- Around line 390-421: The daily summary path acquires
DAILY_NOTIFICATION_LOCK_KEY via acquire_notification_lock but never releases it;
update the acquired-lock branch in the function so that after the summary run
(both on success and on exception) you call
release_notification_lock(DAILY_NOTIFICATION_LOCK_KEY) using
loop.run_until_complete, ensuring the release happens in the same try/finally
that closes db (or in an outer finally) so the lock is always released when
acquired; keep loop.close() after the release and only call release when
lock_acquired is True.
- Around line 543-553: Replace the immediate cancel flow with a graceful wait:
after setting stop_event (app.state.cleanup_stop_event) signal the running
cleanup loop (cleanup_stale_executors) to exit, then await the cleanup_task for
a short timeout (e.g., using asyncio.wait_for(cleanup_task, timeout=...)); only
call cleanup_task.cancel() if the wait_for raises asyncio.TimeoutError, and
still await the task catching asyncio.CancelledError. This ensures
cleanup_stale_executors can finish DB/external deletion work before forced
cancellation.
In `@backend/tests/conftest.py`:
- Around line 165-183: The async_test_db fixture creates an in-memory SQLite
engine but uses the default pooled connections so different connections see
different empty in-memory DBs; modify the create_async_engine call in
async_test_db to use SQLAlchemy's StaticPool (import StaticPool from
sqlalchemy.pool) and pass poolclass=StaticPool (and include
connect_args={"check_same_thread": False} for SQLite) so the same in-memory DB
is shared across connections used by engine.begin() and async_sessionmaker; keep
the rest of the fixture (async_sessionmaker, engine.begin,
Base.metadata.create_all, yield session, engine.dispose) unchanged.
In `@backend/tests/services/test_jobs_cleanup.py`:
- Around line 36-50: The test currently never exercises the "lock held" branch
because cleanup_worker() returns early if _cleanup_recently_executed() is True;
patch app.services.jobs._cleanup_recently_executed (or the module-level function
used by cleanup_worker) to return False within the test context so the code
enters the distributed-lock path, keep the existing patches to
acquire_watchdog_context_async and cleanup_stale_executors (or use their
AsyncMock refs) and then assert that acquire_watchdog_context_async was
awaited/entered (and optionally that cleanup_stale_executors was called) to
validate the lock behavior.
In `@backend/tests/services/test_preserve_executor.py`:
- Around line 124-154: The shared test helper _patch_cleanup_scan currently
patches _scan_candidate_subtasks_batch, _load_tasks_for_cleanup, and
_get_cleanup_subtasks_for_task but not the lookback scan, so
cleanup_stale_executors() may call the real _scan_lookback_subtasks_batch and
hit real async DB code; update _patch_cleanup_scan to also patch
job_service._scan_lookback_subtasks_batch with an AsyncMock returning [] (or
side_effect [[], []] as appropriate) so lookback is stubbed out unless a test
explicitly patches it.
---
Nitpick comments:
In `@backend/app/core/distributed_lock.py`:
- Around line 337-375: In acquire_watchdog_context_async: when acquired is True
but watchdog_task is None (meaning extend_interval_seconds is <=0 or >=
expire_seconds or async_redis_client is None) add a warning log to make the
misconfiguration visible; locate the check that creates watchdog_task (uses
acquired, self.async_redis_client, extend_interval_seconds, expire_seconds) and,
after that block, if acquired and watchdog_task is None, call the module logger
or self.logger to emit a clear warning mentioning lock_name, expire_seconds, and
extend_interval_seconds so users know the extension watchdog was not started.
In `@backend/app/services/workspace_archive/archive_service.py`:
- Around line 378-403: The _try_recover_archive method is directly accessing
archive_storage_service._bucket which couples this code to that module's
internals; add a public helper on archive_storage_service (e.g.
stat_archive(storage_key) that uses the existing client and bucket internally
and returns the object stat or None) and replace the direct call to
archive_storage_service.client.stat_object(archive_storage_service._bucket,
storage_key) with a call to archive_storage_service.stat_archive(storage_key),
then use the returned stat.size as before and keep the same logging/exception
handling in _try_recover_archive.
In `@backend/tests/services/test_preserve_executor.py`:
- Around line 1-1401: The test module is too large (>1000 lines); split it into
smaller modules by logical area: create separate test files for stale cleanup
scans (containing CleanupExecutorTestHelpers and class
TestCleanupStaleExecutorsWithPreserveFlag), manual/task-scoped API cleanup
(TestCleanupTaskExecutorAPI), preserve-label API tests
(TestSetPreserveExecutorAPI), and label consistency tests
(TestPreserveExecutorLabelConsistency). Move related fixtures and helper methods
(e.g., _create_mock_task_resource, _create_mock_subtask, _patch_cleanup_scan)
into the appropriate new modules or a shared conftest/helper module so each new
file imports only what it needs, update imports/patch targets (JobService,
TaskKindsService, TaskResource, Subtask, settings, executor_kinds_service) and
run the test suite to ensure no import/fixture breakage.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 8967efb9-4eda-4c19-be9d-9f6e3b92c585
📒 Files selected for processing (14)
backend/app/api/endpoints/adapter/tasks.pybackend/app/core/cache.pybackend/app/core/config.pybackend/app/core/distributed_lock.pybackend/app/db/session.pybackend/app/services/adapters/executor_job.pybackend/app/services/executor_cleanup_cursor_service.pybackend/app/services/jobs.pybackend/app/services/workspace_archive/archive_service.pybackend/tests/api/endpoints/test_task_cleanup_executor_api.pybackend/tests/conftest.pybackend/tests/services/test_executor_cleanup_progress.pybackend/tests/services/test_jobs_cleanup.pybackend/tests/services/test_preserve_executor.py
| @property | ||
| def async_redis_client(self) -> Optional[AsyncRedis]: | ||
| """Lazy-load async Redis client from settings.""" | ||
| if self._async_redis_client is None: | ||
| try: | ||
| redis_url = getattr(settings, "CELERY_BROKER_URL", None) or getattr( | ||
| settings, "REDIS_URL", None | ||
| ) | ||
| if redis_url: | ||
| self._async_redis_client = AsyncRedis.from_url( | ||
| redis_url, | ||
| decode_responses=True, | ||
| socket_timeout=5, | ||
| socket_connect_timeout=5, | ||
| ) | ||
| except Exception as e: | ||
| logger.warning( | ||
| f"[DistributedLock] Failed to create async Redis client: {e}" | ||
| ) | ||
| self._async_redis_client = None | ||
| return self._async_redis_client |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find all call sites of the new async lock APIs to assess event-loop risk.
rg -nP -C2 '\bacquire_async\b|\brelease_async\b|\bextend_async\b|\bacquire_watchdog_context_async\b' backend
# How is jobs.py executor cleanup loop driven? (celery / asyncio.run / embedded?)
fd -t f 'jobs.py' backend/app/services | xargs -I{} rg -nP -C3 'EXECUTOR_CLEANUP_LOCK_KEY|asyncio\.run|AsyncSessionLocal' {}Repository: wecode-ai/Wegent
Length of output: 5086
🏁 Script executed:
# Understand how jobs.py module is invoked - look for celery tasks or scheduler calls
rg -nP 'def.*cleanup.*executors|cleanup_stale_executors' backend/app/services/jobs.py -A 10
# Search for where this cleanup function is called from
rg -nP 'cleanup_stale_executors|run_scheduled_jobs' backend --type py -B 2
# Check if there's a celery task definition
rg -nP '@.*task|@celery' backend/app/services/jobs.py -A 3Repository: wecode-ai/Wegent
Length of output: 9742
🏁 Script executed:
# Check the distributed_lock singleton implementation
cat -n backend/app/core/distributed_lock.py | head -100
# Look for where distributed_lock is instantiated
rg -nP 'distributed_lock\s*=' backend/app/core/distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 4165
🏁 Script executed:
# Search for celery configuration to understand execution model
fd -t f '*.py' backend/app | xargs rg -l 'Celery|celery\.Celery' | head -5
# Check if there's a scheduler or background task runner
rg -nP 'APScheduler|BackgroundScheduler|interval|cron' backend/app --type py | head -20Repository: wecode-ai/Wegent
Length of output: 2901
🏁 Script executed:
# Get the full function containing the cleanup_stale_executors call
rg -nP 'async def.*\(' backend/app/services/jobs.py | head -20Repository: wecode-ai/Wegent
Length of output: 454
🏁 Script executed:
# Find where jobs.py functions are invoked from - look for imports and calls
rg -nP 'from.*jobs import|import.*jobs' backend/app --type py
# Search for task decorators or scheduled job definitions
rg -nP '@.*task|@app\.on_event|@scheduler' backend/app --type py | head -30Repository: wecode-ai/Wegent
Length of output: 2867
🏁 Script executed:
# Look at the actual context around the cleanup call to find the parent function
sed -n '100,130p' backend/app/services/jobs.pyRepository: wecode-ai/Wegent
Length of output: 1139
🏁 Script executed:
# Find the start_background_jobs function to understand how cleanup_worker is invoked
rg -nP 'def start_background_jobs|async def start_background_jobs' backend/app/services/jobs.py -A 20Repository: wecode-ai/Wegent
Length of output: 832
🏁 Script executed:
# Check main.py to see how background jobs are started
rg -nP 'start_background_jobs' backend/app/main.py -B 3 -A 3Repository: wecode-ai/Wegent
Length of output: 639
🏁 Script executed:
# Look for lifespan or on_event handlers that might control the event loop
rg -nP '@app\.(on_event|lifespan)|lifespan' backend/app/main.py -A 5Repository: wecode-ai/Wegent
Length of output: 950
🏁 Script executed:
# Check if there are other usages of acquire_async, release_async, extend_async, or acquire_watchdog_context_async
rg -nP 'acquire_async|release_async|extend_async|acquire_watchdog_context_async' backend/app --type py | grep -v 'def ' | grep -v '.pyc'Repository: wecode-ai/Wegent
Length of output: 489
🏁 Script executed:
# Look for any test setup that might create multiple event loops
rg -nP 'asyncio\.run|asyncio\.new_event_loop|loop\.run_until_complete|pytest.*asyncio' backend/tests --type py | head -20Repository: wecode-ai/Wegent
Length of output: 1437
🏁 Script executed:
# Check the AsyncSessionLocal to understand if it creates or uses a specific event loop
rg -nP 'AsyncSessionLocal|create_async_engine|async_sessionmaker' backend/app/db/session.py -A 3Repository: wecode-ai/Wegent
Length of output: 1389
🏁 Script executed:
# Look at test files that use the distributed_lock or async lock methods
cat backend/tests/services/test_jobs_cleanup.pyRepository: wecode-ai/Wegent
Length of output: 1549
🏁 Script executed:
# Check if there's any pytest fixture or setup/teardown for distributed_lock
rg -nP 'fixture|setUp|tearDown|conftest' backend/tests --type py | grep -i lockRepository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Look at the actual test file that patches the lock
head -60 backend/tests/services/test_jobs_cleanup.pyRepository: wecode-ai/Wegent
Length of output: 1549
🏁 Script executed:
# Check if there are any other places creating async tasks with redis
rg -nP 'asyncio\.create_task|asyncio\.ensure_future|asyncio\.run' backend/app --type py | grep -v testRepository: wecode-ai/Wegent
Length of output: 6319
🏁 Script executed:
# Verify if distributed_lock is used from request handlers or other contexts
rg -nP 'acquire_watchdog_context_async|acquire_async|release_async|extend_async' backend/app/api --type pyRepository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Check if the Redis client is ever recreated or reset anywhere
rg -nP '_async_redis_client\s*=' backend/app/core/distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 183
🏁 Script executed:
# Final verification: Check if there are any plans to call async lock methods from other contexts
rg -nP 'acquire_async|release_async|extend_async' backend/app --type py | grep -v 'def ' | grep -v distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Verify the ping() check on sync client vs lack on async
sed -n '64,83p' backend/app/core/distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 934
Async Redis client lacks ping verification and has cross-event-loop binding vulnerability.
The _async_redis_client is cached on the module-level singleton, but unlike the sync redis_client property (which validates connectivity via ping()), the async property never verifies that Redis is reachable. If Redis is down at startup, every call to acquire_async / extend_async will silently hang for the full socket_connect_timeout (5 seconds) instead of failing open.
Additionally, redis.asyncio.Redis instances bind their internal connection pool to whatever event loop first issues a command. While current usage is safe—cleanup_worker runs continuously in the single uvicorn event loop—the pattern is fragile. If async lock methods are called from different event loop contexts in the future (e.g., through asyncio.run() or a separate scheduler thread), operations will fail with RuntimeError: Task ... attached to a different loop.
Recommended fixes:
- Add
await client.ping()in the async property initialization (mirroring the sync version) to fail fast if Redis is unreachable. - If multi-loop usage emerges, consider per-call client creation (matching
app/core/cache.pypattern) or cache-per-loop usingid(asyncio.get_running_loop())as key.
🧰 Tools
🪛 Ruff (0.15.10)
[warning] 273-273: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/core/distributed_lock.py` around lines 258 - 278, The async Redis
client property async_redis_client currently caches an AsyncRedis instance via
AsyncRedis.from_url without verifying connectivity or guarding against
event-loop binding; update the async_redis_client getter to create the client,
perform await client.ping() to fail fast if Redis is unreachable (mirroring the
sync redis_client behavior), and then only cache the client if ping succeeds;
additionally, to avoid cross-event-loop RuntimeError, either stop global caching
and create a fresh AsyncRedis via AsyncRedis.from_url per call in
acquire_async/extend_async (matching the per-call pattern in app/core/cache.py)
or implement a cache keyed by the current loop id (use
id(asyncio.get_running_loop())) so the stored client is tied to the calling
event loop.
| from sqlalchemy import create_engine | ||
| from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n backend/app/db/session.py | head -80Repository: wecode-ai/Wegent
Length of output: 3408
🏁 Script executed:
# Check SQLAlchemy version in requirements
fd -e txt -e in | xargs grep -l -i sqlalchemy 2>/dev/null | head -5Repository: wecode-ai/Wegent
Length of output: 85
🏁 Script executed:
# Check requirements files for SQLAlchemy version
rg -i "sqlalchemy" --type txt -A 1 -B 1Repository: wecode-ai/Wegent
Length of output: 182
🏁 Script executed:
# Check if there are any other requirements files with SQLAlchemy
find . -name "requirements*.txt" -o -name "pyproject.toml" -o -name "setup.py" | xargs grep -l -i sqlalchemy 2>/dev/nullRepository: wecode-ai/Wegent
Length of output: 224
🏁 Script executed:
# Check SQLAlchemy version in backend pyproject.toml
cat backend/pyproject.toml | grep -A 2 -B 2 sqlalchemyRepository: wecode-ai/Wegent
Length of output: 450
🏁 Script executed:
# Check if asyncmy and aiosqlite are listed as dependencies
rg "asyncmy|aiosqlite" backend/pyproject.tomlRepository: wecode-ai/Wegent
Length of output: 81
🏁 Script executed:
# Check if aiosqlite is a dependency anywhere
rg "aiosqlite" . --type toml --type txtRepository: wecode-ai/Wegent
Length of output: 112
🌐 Web query:
SQLAlchemy 2.0 make_url set drivername render_as_string
💡 Result:
In SQLAlchemy 2.0, use make_url to create a URL object from a string, URL.create for programmatic construction, URL.set to modify an existing URL object (including setting drivername), and URL.render_as_string to get the string representation. Example workflow: from sqlalchemy.engine import make_url, URL # From string url = make_url("postgresql://user:pass@localhost/db") # Create new url2 = URL.create(drivername="mysql+pymysql", host="example.com", database="mydb") # Set drivername on existing url3 = url.set(drivername="postgresql+psycopg2") # Render as string url_str = url3.render_as_string(hide_password=False) print(url_str) # postgresql+psycopg2://user:pass@localhost/db URL objects are immutable; set returns a new object with modifications. drivername is the dialect+driver part (e.g., "postgresql+psycopg2").
Citations:
- 1: https://www.sqlalchemy.org/docs/stable/core/engines.html
- 2: https://docs.sqlalchemy.org/en/stable/core/engines.html
- 3: https://github.com/zzzeek/sqlalchemy/blob/master/lib/sqlalchemy/engine/url.py
- 4: https://pydoc.dev/sqlalchemy/latest/sqlalchemy.engine.url.URL.html
- 5: https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/engine/url.py
🏁 Script executed:
# Check where _get_async_database_url is used
rg "_get_async_database_url" backend/Repository: wecode-ai/Wegent
Length of output: 191
Use SQLAlchemy URL parsing for async driver conversion.
The fallback on line 74 creates invalid dialect strings. For example, a URL like mysql+mysqldb://... becomes mysql+asyncmy+mysqldb://..., which SQLAlchemy cannot parse. Use make_url() to properly extract and set the driver name.
Safer conversion
from sqlalchemy import create_engine
+from sqlalchemy.engine import make_url
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
@@
def _get_async_database_url() -> str:
"""Convert sync DATABASE_URL to async driver URL."""
- url = SQLALCHEMY_DATABASE_URL
- if "pymysql" in url:
- return url.replace("pymysql", "asyncmy")
- elif url.startswith("mysql://"):
- return url.replace("mysql://", "mysql+asyncmy://")
- elif "asyncmy" in url:
- return url
- elif url.startswith("sqlite"):
- return url.replace("sqlite://", "sqlite+aiosqlite://")
- return url.replace("mysql+", "mysql+asyncmy+", 1)
+ url = make_url(SQLALCHEMY_DATABASE_URL)
+ if url.drivername.startswith("mysql"):
+ url = url.set(drivername="mysql+asyncmy")
+ elif url.drivername.startswith("sqlite"):
+ url = url.set(drivername="sqlite+aiosqlite")
+ return url.render_as_string(hide_password=False)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/db/session.py` around lines 8 - 9, The URL-to-async conversion
currently concatenates strings and produces invalid dialects (e.g.,
mysql+mysqldb -> mysql+asyncmy+mysqldb); instead parse the DB URL with
sqlalchemy.engine.make_url (or make_url) and rebuild the drivername properly:
extract the dialect and existing driver from the parsed URL, replace the sync
driver with the intended async driver (keeping the dialect prefix), assign the
new drivername back to the parsed URL, and pass that URL to
create_async_engine/AsyncEngine/AsyncSession creation code (the code around
AsyncEngine/create_async_engine in session.py). Update the conversion logic to
use make_url so SQLAlchemy receives a valid URL rather than a simple string
concat.
| scanned_batches += 1 | ||
| total_scanned += len(scanned_subtasks) | ||
| last_id = scanned_subtasks[-1].id | ||
| await executor_cleanup_cursor_service.advance_cursor( | ||
| db, | ||
| last_scanned_subtask_id=last_id, | ||
| ) | ||
| batch_candidates = self._filter_scanned_subtasks( | ||
| scanned_subtasks, | ||
| chat_cutoff=cutoff, | ||
| ) | ||
| ( | ||
| batch_valid_count, | ||
| batch_deleted_count, | ||
| batch_filter_stats, | ||
| ) = await self._process_cleanup_batch( | ||
| db=db, | ||
| candidates=batch_candidates, | ||
| chat_cutoff=cutoff, | ||
| ) | ||
| total_valid += batch_valid_count | ||
| total_filter_stats = self._merge_cleanup_filter_stats( | ||
| total_filter_stats, batch_filter_stats | ||
| ) | ||
| total_deleted += batch_deleted_count |
There was a problem hiding this comment.
Advance the cursor only after the batch is processed.
The cursor is persisted before _process_cleanup_batch() runs. If task loading/deletion processing fails after Line 130, those subtasks are considered scanned and may never be retried by the primary cursor path.
Move cursor advancement after successful batch processing
scanned_batches += 1
total_scanned += len(scanned_subtasks)
last_id = scanned_subtasks[-1].id
- await executor_cleanup_cursor_service.advance_cursor(
- db,
- last_scanned_subtask_id=last_id,
- )
batch_candidates = self._filter_scanned_subtasks(
scanned_subtasks,
chat_cutoff=cutoff,
)
@@
total_filter_stats = self._merge_cleanup_filter_stats(
total_filter_stats, batch_filter_stats
)
total_deleted += batch_deleted_count
+ await executor_cleanup_cursor_service.advance_cursor(
+ db,
+ last_scanned_subtask_id=last_id,
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| scanned_batches += 1 | |
| total_scanned += len(scanned_subtasks) | |
| last_id = scanned_subtasks[-1].id | |
| await executor_cleanup_cursor_service.advance_cursor( | |
| db, | |
| last_scanned_subtask_id=last_id, | |
| ) | |
| batch_candidates = self._filter_scanned_subtasks( | |
| scanned_subtasks, | |
| chat_cutoff=cutoff, | |
| ) | |
| ( | |
| batch_valid_count, | |
| batch_deleted_count, | |
| batch_filter_stats, | |
| ) = await self._process_cleanup_batch( | |
| db=db, | |
| candidates=batch_candidates, | |
| chat_cutoff=cutoff, | |
| ) | |
| total_valid += batch_valid_count | |
| total_filter_stats = self._merge_cleanup_filter_stats( | |
| total_filter_stats, batch_filter_stats | |
| ) | |
| total_deleted += batch_deleted_count | |
| scanned_batches += 1 | |
| total_scanned += len(scanned_subtasks) | |
| last_id = scanned_subtasks[-1].id | |
| batch_candidates = self._filter_scanned_subtasks( | |
| scanned_subtasks, | |
| chat_cutoff=cutoff, | |
| ) | |
| ( | |
| batch_valid_count, | |
| batch_deleted_count, | |
| batch_filter_stats, | |
| ) = await self._process_cleanup_batch( | |
| db=db, | |
| candidates=batch_candidates, | |
| chat_cutoff=cutoff, | |
| ) | |
| total_valid += batch_valid_count | |
| total_filter_stats = self._merge_cleanup_filter_stats( | |
| total_filter_stats, batch_filter_stats | |
| ) | |
| total_deleted += batch_deleted_count | |
| await executor_cleanup_cursor_service.advance_cursor( | |
| db, | |
| last_scanned_subtask_id=last_id, | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/adapters/executor_job.py` around lines 127 - 151, The
cursor is advanced too early: currently
executor_cleanup_cursor_service.advance_cursor(...) is called right after
computing last_id from scanned_subtasks but before calling
_process_cleanup_batch(...); move the advance_cursor call to after
_process_cleanup_batch returns successfully (after
batch_valid_count/batch_deleted_count/batch_filter_stats are obtained) so that
only fully processed subtasks mark the cursor; keep using last_id computed from
scanned_subtasks and only call advance_cursor when _process_cleanup_batch
completes without raising.
| result = await db.execute( | ||
| select(Subtask) | ||
| .filter( | ||
| Subtask.status.in_( | ||
| [ | ||
| SubtaskStatus.PENDING, | ||
| SubtaskStatus.COMPLETED, | ||
| SubtaskStatus.FAILED, | ||
| SubtaskStatus.CANCELLED, | ||
| ] | ||
| ), | ||
| Subtask.created_at > lookback_start, | ||
| Subtask.created_at <= cutoff, | ||
| Subtask.executor_name.isnot(None), | ||
| Subtask.executor_name != "", | ||
| Subtask.executor_deleted_at == False, | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
wc -l backend/app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent
Length of output: 109
🏁 Script executed:
sed -n '263,279p' backend/app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent
Length of output: 696
🏁 Script executed:
sed -n '531,537p' backend/app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent
Length of output: 338
🏁 Script executed:
sed -n '714,720p' backend/app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent
Length of output: 354
🏁 Script executed:
# Search for all occurrences of == False in SQLAlchemy filter contexts
rg "== False" backend/app/services/adapters/executor_job.py -nRepository: wecode-ai/Wegent
Length of output: 241
🏁 Script executed:
# Search for existing uses of .is_() in the codebase to see if it's already used
rg "\.is_\(" backend/app/ -t py -B 1 -A 1 | head -40Repository: wecode-ai/Wegent
Length of output: 2510
🌐 Web query:
SQLAlchemy .is_() method boolean comparison SQLAlchemy ORM boolean predicates
💡 Result:
The SQLAlchemy .is_ method is a ColumnOperator that generates SQL "IS" comparisons, primarily for NULL checks (e.g., column.is_(None) renders as column IS NULL). It is automatically used when == None is written, but Python's is keyword cannot be overloaded, so column is None performs Python identity comparison on the ColumnElement object and always evaluates to False in filters. For boolean columns in SQLAlchemy ORM predicates: - Use column == True or simply column (implicitly true): renders column = true (or dialect-appropriate like column = 1). - Use column == False: renders column = false. - Use column.is_(true) or column.is_(false) from sqlalchemy import true, false: renders column IS true/false (dialect-dependent; some backends prefer this for native boolean constants). - Use ~column for negation: renders NOT column. - Avoid column is True/False: this does Python-level identity check on the ColumnElement, not SQL. Examples in ORM queries: from sqlalchemy import true, false session.query(Model).filter(Model.bool_col == True).all # column = true session.query(Model).filter(Model.bool_col).all # same as above session.query(Model).filter(Model.bool_col.is_(false)).all # column IS false session.query(Model).filter(~Model.bool_col).all # NOT bool_col Both == and .is_ work for booleans but produce different SQL (= vs IS), with == more common. Use true/false constants for dialect-portable boolean literals. Official docs confirm .is_ for identity/NULL, while booleans use overloaded operators.
Citations:
- 1: https://docs.sqlalchemy.org/en/14/core/operators.html
- 2: http://docs.sqlalchemy.org/en/latest/core/operators.html
- 3: https://www.sqlalchemy.org/docs/07/core/operators.html
- 4: https://stackoverflow.com/questions/48274553/sqlalchemy-core-select-where-condition-contains-boolean-expression-is-false
- 5: https://stackoverflow.com/questions/62451672/sqlalchemy-cant-get-data-with-boolean-field-in-filter-option-specified-with-is
- 6: https://stackoverflow.com/questions/18998010/flake8-complains-on-boolean-comparison-in-filter-clause
- 7: https://stackoverflow.com/questions/31811119/boolean-field-query-with-sqlalchemy
- 8: https://docs.sqlalchemy.org/core/sqlelement.html
- 9: https://docs.sqlalchemy.org/en/14/core/sqlelement.html
🌐 Web query:
Ruff linter rule SQLAlchemy == False comparison
💡 Result:
The Ruff linter rule flagging SQLAlchemy == False comparisons is E712 (true-false-comparison). This rule detects == True or == False and suggests using implicit truthiness (if cond: or if not cond:) or is/is not. However, in SQLAlchemy (e.g., Query.filter(column == False)), == generates SQL = operators, while is/is not would use SQL IS/IS NOT, which behave differently (especially with NULLs). Ruff's auto-fix to is/is not changes semantics and can break queries. SQLAlchemy provides safe alternatives like column.is_(False) or column == sqlalchemy.false, but Ruff does not auto-convert to these and flags the == False pattern. Known issue since 2023 (GitHub #4560); no automatic exemption for SQLAlchemy contexts. Workarounds: disable E712 per-file (# noqa: E712), in config (ignore = ["E712"]), or use # pyright: ignore if combined with other tools. Docs warn the fix is unsafe for libraries overriding eq like SQLAlchemy.
Citations:
- 1: https://docs.astral.sh/ruff/rules/type-comparison/
- 2: https://docs.astral.sh/ruff/rules/true-false-comparison/
- 3: Fix (E712) changing
==/!=tois/is notis not correct for some types astral-sh/ruff#4560 - 4: Fix (E712) changing
==/!=tois/is notis not correct for some types astral-sh/ruff#4560 - 5: [ruff --fix] (v0.0.239) When working with numpy arrays
mask == Trueis not the same asmask is Trueastral-sh/ruff#2443
Use .is_(False) to satisfy Ruff's boolean comparison linting.
Ruff flags == False comparisons with rule E712. Replace with .is_(False) to align with the codebase's established pattern (used in group_service.py, work_queue_service.py, and base_service.py).
SQLAlchemy boolean comparison
- Subtask.executor_deleted_at == False,
+ Subtask.executor_deleted_at.is_(False),Also applies to lines 536 and 718.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| result = await db.execute( | |
| select(Subtask) | |
| .filter( | |
| Subtask.status.in_( | |
| [ | |
| SubtaskStatus.PENDING, | |
| SubtaskStatus.COMPLETED, | |
| SubtaskStatus.FAILED, | |
| SubtaskStatus.CANCELLED, | |
| ] | |
| ), | |
| Subtask.created_at > lookback_start, | |
| Subtask.created_at <= cutoff, | |
| Subtask.executor_name.isnot(None), | |
| Subtask.executor_name != "", | |
| Subtask.executor_deleted_at == False, | |
| ) | |
| result = await db.execute( | |
| select(Subtask) | |
| .filter( | |
| Subtask.status.in_( | |
| [ | |
| SubtaskStatus.PENDING, | |
| SubtaskStatus.COMPLETED, | |
| SubtaskStatus.FAILED, | |
| SubtaskStatus.CANCELLED, | |
| ] | |
| ), | |
| Subtask.created_at > lookback_start, | |
| Subtask.created_at <= cutoff, | |
| Subtask.executor_name.isnot(None), | |
| Subtask.executor_name != "", | |
| Subtask.executor_deleted_at.is_(False), | |
| ) |
🧰 Tools
🪛 Ruff (0.15.10)
[error] 278-278: Avoid equality comparisons to False; use not Subtask.executor_deleted_at: for false checks
Replace with not Subtask.executor_deleted_at
(E712)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/adapters/executor_job.py` around lines 263 - 279,
Replace direct boolean equality checks against False with SQLAlchemy's is_
comparison: change occurrences of Subtask.executor_deleted_at == False to
Subtask.executor_deleted_at.is_(False) in the query used with
db.execute/select(Subtask) (the block assigning result). Also update the same
pattern at the other two occurrences referenced (around the logic near lines
shown — the other similar Subtask queries at the later usages) so all checks use
.is_(False) instead of == False.
| def _filter_scanned_subtasks( | ||
| self, scanned_subtasks: List[Subtask], *, chat_cutoff: datetime | ||
| ) -> List[Subtask]: | ||
| """Filter raw scanned subtasks down to cleanup-eligible subtask candidates.""" | ||
| return [ | ||
| subtask | ||
| for subtask in scanned_subtasks | ||
| if subtask.status | ||
| in [ | ||
| SubtaskStatus.PENDING, | ||
| SubtaskStatus.COMPLETED, | ||
| SubtaskStatus.FAILED, | ||
| SubtaskStatus.CANCELLED, | ||
| ] | ||
| and isinstance(subtask.updated_at, datetime) | ||
| and subtask.updated_at <= chat_cutoff | ||
| and bool(subtask.executor_name) | ||
| and not self._is_device_executor_name(subtask.executor_name) | ||
| and not bool(subtask.executor_deleted_at) |
There was a problem hiding this comment.
Require executor_namespace wherever cleanup candidates are selected.
The scheduled cleanup docstring says both executor name and namespace must be non-empty, but the filters only require executor_name. That can pass None/"" namespaces into delete_executor_task_async().
Add namespace filtering
and subtask.updated_at <= chat_cutoff
and bool(subtask.executor_name)
+ and bool(subtask.executor_namespace)
and not self._is_device_executor_name(subtask.executor_name)
and not bool(subtask.executor_deleted_at)
]
@@
Subtask.task_id == task_id,
Subtask.executor_name.isnot(None),
Subtask.executor_name != "",
+ Subtask.executor_namespace.isnot(None),
+ Subtask.executor_namespace != "",
Subtask.executor_deleted_at == False,
)Also applies to: 527-539
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/adapters/executor_job.py` around lines 285 - 303, The
filter in _filter_scanned_subtasks currently allows subtasks with an empty or
None executor_namespace even though scheduled cleanup requires both
executor_name and executor_namespace; update the list comprehension to also
require a truthy subtask.executor_namespace (and keep the other checks: status
in {PENDING, COMPLETED, FAILED, CANCELLED}, isinstance(updated_at, datetime) and
updated_at <= chat_cutoff, truthy executor_name, not
_is_device_executor_name(executor_name), and not executor_deleted_at) so no
subtasks with missing namespaces are forwarded to delete_executor_task_async;
apply the same executor_namespace truthy check to the analogous filter block
referenced around lines 527-539.
| @pytest.fixture(scope="function") | ||
| async def async_test_db(): | ||
| """ | ||
| Async test DB session for testing async service methods. | ||
| Uses an independent in-memory SQLite database with its own tables. | ||
| Data setup should be done via this session (not test_db). | ||
| """ | ||
| engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False) | ||
|
|
||
| async with engine.begin() as conn: | ||
| await conn.run_sync(Base.metadata.create_all) | ||
|
|
||
| factory = async_sessionmaker(engine, expire_on_commit=False) | ||
|
|
||
| async with factory() as session: | ||
| yield session | ||
|
|
||
| await engine.dispose() | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
SQLAlchemy create_async_engine sqlite aiosqlite in-memory StaticPool connection sharing
💡 Result:
To create an async SQLAlchemy engine using aiosqlite with an in-memory SQLite database and StaticPool for connection sharing: Use create_async_engine with the sqlite+aiosqlite:///:memory: URL and specify poolclass=StaticPool from sqlalchemy.pool. For SQLite in-memory databases, StaticPool provides exactly one shared connection across all requests, enabling data sharing which is otherwise impossible with standard :memory: (each connection gets a private DB). Example code: from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.pool import StaticPool engine = create_async_engine( "sqlite+aiosqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False} # Disable SQLite thread check for shared use ) Key points from official docs: - Standard :memory: creates isolated DB per connection; StaticPool + shared URI or simple :memory: with StaticPool enables sharing via single connection. - aiosqlite works with SQLAlchemy async via create_async_engine; pooling uses AsyncAdaptedQueuePool by default (recent versions), but StaticPool can be specified explicitly. - check_same_thread=False is required for sharing the single connection safely. - For true shared in-memory across "connections" (named shared cache), use file:dbname?mode=memory&cache=shared URI, but StaticPool is simpler for single shared connection. Usage with session: from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.asyncio import AsyncSession AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with AsyncSessionLocal as session: # operations here await session.commit This setup ensures all async sessions share the same underlying SQLite connection and in-memory database state. Note: SQLite has writer starvation limits in concurrent async use; suitable for testing/low-concurrency.
Citations:
- 1: http://docs.sqlalchemy.org/en/latest/dialects/sqlite.html
- 2: http://docs.sqlalchemy.org/en/latest/core/pooling.html
- 3: https://www.sqlalchemy.org/docs/21/core/pooling.html
- 4: https://sqlalchemy-sqlalchemy.mintlify.app/core/pooling
- 5: https://gist.github.com/wynemo/81c3e0907c5ae0737f05debac39a1b53
- 6: https://stackoverflow.com/questions/27910829/sqlalchemy-and-sqlite-shared-cache
🏁 Script executed:
cat -n backend/tests/conftest.py | sed -n '61,77p;165,183p'Repository: wecode-ai/Wegent
Length of output: 1597
In-memory async SQLite engine needs StaticPool.
sqlite+aiosqlite:///:memory: with the default AsyncAdaptedQueuePool gives each pooled connection its own isolated in-memory database. Within a single test, when engine.begin() creates the schema on one connection and factory() later checks out a different connection from the pool, the second connection has no tables, causing flaky no such table errors. The sync test_engine fixture correctly uses StaticPool for this reason (line 77) — the async fixture should do the same.
Proposed fix
- engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+ from sqlalchemy.pool import StaticPool
+
+ engine = create_async_engine(
+ "sqlite+aiosqlite:///:memory:",
+ echo=False,
+ poolclass=StaticPool,
+ connect_args={"check_same_thread": False},
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @pytest.fixture(scope="function") | |
| async def async_test_db(): | |
| """ | |
| Async test DB session for testing async service methods. | |
| Uses an independent in-memory SQLite database with its own tables. | |
| Data setup should be done via this session (not test_db). | |
| """ | |
| engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False) | |
| async with engine.begin() as conn: | |
| await conn.run_sync(Base.metadata.create_all) | |
| factory = async_sessionmaker(engine, expire_on_commit=False) | |
| async with factory() as session: | |
| yield session | |
| await engine.dispose() | |
| `@pytest.fixture`(scope="function") | |
| async def async_test_db(): | |
| """ | |
| Async test DB session for testing async service methods. | |
| Uses an independent in-memory SQLite database with its own tables. | |
| Data setup should be done via this session (not test_db). | |
| """ | |
| from sqlalchemy.pool import StaticPool | |
| engine = create_async_engine( | |
| "sqlite+aiosqlite:///:memory:", | |
| echo=False, | |
| poolclass=StaticPool, | |
| connect_args={"check_same_thread": False}, | |
| ) | |
| async with engine.begin() as conn: | |
| await conn.run_sync(Base.metadata.create_all) | |
| factory = async_sessionmaker(engine, expire_on_commit=False) | |
| async with factory() as session: | |
| yield session | |
| await engine.dispose() |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/tests/conftest.py` around lines 165 - 183, The async_test_db fixture
creates an in-memory SQLite engine but uses the default pooled connections so
different connections see different empty in-memory DBs; modify the
create_async_engine call in async_test_db to use SQLAlchemy's StaticPool (import
StaticPool from sqlalchemy.pool) and pass poolclass=StaticPool (and include
connect_args={"check_same_thread": False} for SQLite) so the same in-memory DB
is shared across connections used by engine.begin() and async_sessionmaker; keep
the rest of the fixture (async_sessionmaker, engine.begin,
Base.metadata.create_all, yield session, engine.dispose) unchanged.
| with ( | ||
| patch( | ||
| "app.services.jobs.distributed_lock.acquire_watchdog_context_async", | ||
| return_value=_lock_context(False), | ||
| create=True, | ||
| ), | ||
| patch( | ||
| "app.services.jobs.job_service.cleanup_stale_executors", | ||
| new_callable=AsyncMock, | ||
| ) as cleanup_mock, | ||
| patch("app.services.jobs.AsyncSessionLocal") as session_local, | ||
| patch("app.services.jobs.settings") as mock_settings, | ||
| ): | ||
| mock_settings.TASK_EXECUTOR_CLEANUP_INTERVAL_SECONDS = 0.01 | ||
| await _run_worker() |
There was a problem hiding this comment.
Force the worker past the recent-cleanup guard before asserting lock behavior.
cleanup_worker() checks _cleanup_recently_executed() before entering the distributed-lock context, so this test can pass without actually testing the “lock held” branch. Patch that guard to False and optionally assert the lock mock was reached.
Suggested test hardening
with (
+ patch(
+ "app.services.jobs._cleanup_recently_executed",
+ new_callable=AsyncMock,
+ return_value=False,
+ ),
patch(
"app.services.jobs.distributed_lock.acquire_watchdog_context_async",
return_value=_lock_context(False),
create=True,
- ),
+ ) as acquire_lock_mock,
patch(
"app.services.jobs.job_service.cleanup_stale_executors",
new_callable=AsyncMock,
) as cleanup_mock,
@@
mock_settings.TASK_EXECUTOR_CLEANUP_INTERVAL_SECONDS = 0.01
await _run_worker()
+ acquire_lock_mock.assert_called()
cleanup_mock.assert_not_called()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/tests/services/test_jobs_cleanup.py` around lines 36 - 50, The test
currently never exercises the "lock held" branch because cleanup_worker()
returns early if _cleanup_recently_executed() is True; patch
app.services.jobs._cleanup_recently_executed (or the module-level function used
by cleanup_worker) to return False within the test context so the code enters
the distributed-lock path, keep the existing patches to
acquire_watchdog_context_async and cleanup_stale_executors (or use their
AsyncMock refs) and then assert that acquire_watchdog_context_async was
awaited/entered (and optionally that cleanup_stale_executors was called) to
validate the lock behavior.
| @contextmanager | ||
| def _patch_cleanup_scan(self, job_service, subtasks, task_map): | ||
| """Patch the scheduled cleanup scan helpers for one logical batch.""" | ||
| with ExitStack() as stack: | ||
| stack.enter_context( | ||
| patch.object( | ||
| job_service, | ||
| "_scan_candidate_subtasks_batch", | ||
| new_callable=AsyncMock, | ||
| side_effect=[subtasks, []], | ||
| ) | ||
| ) | ||
| stack.enter_context( | ||
| patch.object( | ||
| job_service, | ||
| "_load_tasks_for_cleanup", | ||
| new_callable=AsyncMock, | ||
| return_value=task_map, | ||
| ) | ||
| ) | ||
| stack.enter_context( | ||
| patch.object( | ||
| job_service, | ||
| "_get_cleanup_subtasks_for_task", | ||
| new_callable=AsyncMock, | ||
| side_effect=lambda _db, task_id: [ | ||
| subtask for subtask in subtasks if subtask.task_id == task_id | ||
| ], | ||
| ) | ||
| ) | ||
| yield |
There was a problem hiding this comment.
Patch the lookback scan in the shared cleanup helper.
After the primary scan returns [], cleanup_stale_executors() still runs _scan_lookback_subtasks_batch() when the deleted target is not reached. Tests using mock_db can therefore fall through into real async DB code. Patch lookback to [] here unless a test explicitly wants that path.
Keep helper-isolated tests out of the real lookback scan
stack.enter_context(
patch.object(
job_service,
"_get_cleanup_subtasks_for_task",
new_callable=AsyncMock,
@@
],
)
)
+ stack.enter_context(
+ patch.object(
+ job_service,
+ "_scan_lookback_subtasks_batch",
+ new_callable=AsyncMock,
+ return_value=[],
+ create=True,
+ )
+ )
yield🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/tests/services/test_preserve_executor.py` around lines 124 - 154, The
shared test helper _patch_cleanup_scan currently patches
_scan_candidate_subtasks_batch, _load_tasks_for_cleanup, and
_get_cleanup_subtasks_for_task but not the lookback scan, so
cleanup_stale_executors() may call the real _scan_lookback_subtasks_batch and
hit real async DB code; update _patch_cleanup_scan to also patch
job_service._scan_lookback_subtasks_batch with an AsyncMock returning [] (or
side_effect [[], []] as appropriate) so lookback is stubbed out unless a test
explicitly patches it.
| @pytest.mark.skip(reason="Test needs to be updated for new cleanup implementation") | ||
| async def test_cleanup_logs_filter_stats_when_no_valid_candidates( | ||
| self, job_service, mock_db | ||
| ): | ||
| """Test cleanup summary logs include filter reason counts.""" | ||
| pass |
There was a problem hiding this comment.
Update or remove the skipped filter-stats test.
This PR adds filter counters/logging, but the dedicated assertion is skipped with “needs to be updated”. That leaves the new cleanup observability path unprotected.
Do you want me to generate an updated test for the new batched cleanup implementation?
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…jobs.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Refactor cleanup_worker to async with distributed lock coordination and Redis cursor-based deduplication across pods. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
backend/app/services/jobs.py (1)
237-247:⚠️ Potential issue | 🟠 MajorLet cleanup exit cooperatively before forcing cancellation.
Setting
cleanup_stop_eventis enough for the worker to stop after the current iteration. Immediatecleanup_task.cancel()can still interruptcleanup_stale_executors()mid-flight, including between external deletion and DB state updates.🛠️ Proposed graceful shutdown flow
if stop_event: stop_event.set() if cleanup_task: - cleanup_task.cancel() try: - await cleanup_task - except asyncio.CancelledError: - pass + await asyncio.wait_for(cleanup_task, timeout=5.0) + except asyncio.TimeoutError: + cleanup_task.cancel() + try: + await cleanup_task + except asyncio.CancelledError: + pass🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/jobs.py` around lines 237 - 247, Set the cleanup_stop_event and allow cleanup_task to finish cooperatively before forcing cancellation: set getattr(app.state, "cleanup_stop_event"), then await the existing cleanup_task with a short configurable grace period (e.g., via asyncio.wait_for or asyncio.shield) so cleanup_stale_executors() can finish its current iteration and complete DB/external deletion steps; only call cleanup_task.cancel() if the task does not finish within the grace timeout, and handle asyncio.TimeoutError and asyncio.CancelledError appropriately. Use the identifiers cleanup_stop_event, cleanup_task, and cleanup_stale_executors to locate and update the shutdown logic.
🧹 Nitpick comments (1)
backend/app/services/jobs.py (1)
101-101: Add type hints to the changed public lifecycle functions.
cleanup_worker,start_background_jobs, andstop_background_jobsare public module functions but are missing complete annotations.♻️ Proposed annotation cleanup
+from fastapi import FastAPI + -async def cleanup_worker(stop_event: asyncio.Event): +async def cleanup_worker(stop_event: asyncio.Event) -> None: ... -def start_background_jobs(app): +def start_background_jobs(app: FastAPI) -> None: ... -async def stop_background_jobs(app): +async def stop_background_jobs(app: FastAPI) -> None:As per coding guidelines, Use type hints for all public functions and classes in Python.
Also applies to: 199-199, 230-230
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/jobs.py` at line 101, Add explicit type hints to the three public lifecycle functions: annotate cleanup_worker(stop_event: asyncio.Event) with the asyncio.Event parameter type and a return type of None, and similarly add full parameter and return annotations for start_background_jobs and stop_background_jobs (e.g., explicit app/context types if they accept the application or shutdown event, and -> None for their return). Import any required types (asyncio.Event, typing.Optional, or the app type) at the top and ensure the function signatures for cleanup_worker, start_background_jobs, and stop_background_jobs include these annotations and updated return types.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/services/jobs.py`:
- Line 230: The shutdown call to the async function stop_background_jobs(app) is
not awaited inside the async lifespan context manager, creating an un-awaited
coroutine and skipping cleanup; update the call site in the async lifespan
shutdown phase to use await stop_background_jobs(app) so the coroutine is
executed and background jobs are properly stopped.
- Around line 113-124: The current block uses
distributed_lock.acquire_watchdog_context_async (and thus acquire_async) which
is fail-open when Redis errors occur; change executor cleanup to fail-closed by
performing a Redis health check before running cleanup: before entering the
async with acquire_watchdog_context_async(EXECUTOR_CLEANUP_LOCK_KEY,...), call
the lock/redis health method (e.g., distributed_lock.ping() or a provided health
check on distributed_lock) and if the health check fails, log and skip cleanup;
alternatively, if acquire_watchdog_context_async supports a strict/fail_closed
flag, pass that option; ensure you still use EXECUTOR_CLEANUP_LOCK_KEY and only
call job_service.cleanup_stale_executors(db) when the health check passes and
the lock was truly acquired.
---
Duplicate comments:
In `@backend/app/services/jobs.py`:
- Around line 237-247: Set the cleanup_stop_event and allow cleanup_task to
finish cooperatively before forcing cancellation: set getattr(app.state,
"cleanup_stop_event"), then await the existing cleanup_task with a short
configurable grace period (e.g., via asyncio.wait_for or asyncio.shield) so
cleanup_stale_executors() can finish its current iteration and complete
DB/external deletion steps; only call cleanup_task.cancel() if the task does not
finish within the grace timeout, and handle asyncio.TimeoutError and
asyncio.CancelledError appropriately. Use the identifiers cleanup_stop_event,
cleanup_task, and cleanup_stale_executors to locate and update the shutdown
logic.
---
Nitpick comments:
In `@backend/app/services/jobs.py`:
- Line 101: Add explicit type hints to the three public lifecycle functions:
annotate cleanup_worker(stop_event: asyncio.Event) with the asyncio.Event
parameter type and a return type of None, and similarly add full parameter and
return annotations for start_background_jobs and stop_background_jobs (e.g.,
explicit app/context types if they accept the application or shutdown event, and
-> None for their return). Import any required types (asyncio.Event,
typing.Optional, or the app type) at the top and ensure the function signatures
for cleanup_worker, start_background_jobs, and stop_background_jobs include
these annotations and updated return types.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ad646318-b4f4-4235-bea4-793b3614c950
📒 Files selected for processing (1)
backend/app/services/jobs.py
| async with distributed_lock.acquire_watchdog_context_async( | ||
| EXECUTOR_CLEANUP_LOCK_KEY, | ||
| expire_seconds=300, | ||
| extend_interval_seconds=60, | ||
| ) as acquired: | ||
| if not acquired: | ||
| logger.info( | ||
| "[job] Another instance is executing executor cleanup, skipping this execution" | ||
| ) | ||
| else: | ||
| async with AsyncSessionLocal() as db: | ||
| await job_service.cleanup_stale_executors(db) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify distributed lock fail-open behavior and executor cleanup lock usage.
rg -nP -C4 'def acquire_async|return True|acquire_watchdog_context_async|EXECUTOR_CLEANUP_LOCK_KEY' \
--glob '*.py'Repository: wecode-ai/Wegent
Length of output: 50373
🏁 Script executed:
#!/bin/bash
# Find and examine the distributed lock implementation
find . -name "distributed_lock.py" -type fRepository: wecode-ai/Wegent
Length of output: 144
🏁 Script executed:
#!/bin/bash
# Find and examine the jobs.py file in backend services
find . -path "*/backend/app/services/jobs.py" -type fRepository: wecode-ai/Wegent
Length of output: 90
🏁 Script executed:
#!/bin/bash
# Search for acquire_async and acquire_watchdog_context_async definitions
rg -n "def acquire_async|def acquire_watchdog_context_async" --glob "*.py"Repository: wecode-ai/Wegent
Length of output: 272
🏁 Script executed:
#!/bin/bash
# Read the distributed_lock.py implementation of acquire_async
sed -n '280,310p' backend/app/core/distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 1255
🏁 Script executed:
#!/bin/bash
# Read the acquire_watchdog_context_async implementation
sed -n '338,380p' backend/app/core/distributed_lock.pyRepository: wecode-ai/Wegent
Length of output: 1444
🏁 Script executed:
#!/bin/bash
# Read the cleanup_worker in jobs.py around lines 113-124
sed -n '100,130p' backend/app/services/jobs.pyRepository: wecode-ai/Wegent
Length of output: 1192
🏁 Script executed:
#!/bin/bash
# Find the cleanup_stale_executors function definition
rg -n "def cleanup_stale_executors" --glob "*.py" -A 20Repository: wecode-ai/Wegent
Length of output: 2099
🏁 Script executed:
#!/bin/bash
# Get more context on _cleanup_recently_executed to understand full cleanup flow
rg -n "async def _cleanup_recently_executed" backend/app/services/jobs.py -A 15Repository: wecode-ai/Wegent
Length of output: 817
🏁 Script executed:
#!/bin/bash
# Check if there are any tests for the distributed lock fail-open behavior
rg -n "acquire_async|acquire_watchdog_context_async" backend/app/core/distributed_lock.py -B 2 -A 15 | head -80Repository: wecode-ai/Wegent
Length of output: 1843
🏁 Script executed:
#!/bin/bash
# Understand the scope of cleanup_stale_executors - check for external API calls
sed -n '86,150p' backend/app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent
Length of output: 2789
🏁 Script executed:
#!/bin/bash
# Check if there are any tests that demonstrate the intended fail-open behavior
rg -n "async_redis_client is None|Async Redis not available" backend/app/core/distributed_lock.py -B 2 -A 2Repository: wecode-ai/Wegent
Length of output: 1164
🏁 Script executed:
#!/bin/bash
# Look for executor deletion API call to check idempotency
rg -n "delete|DELETE" backend/app/services/adapters/executor_job.py | head -20Repository: wecode-ai/Wegent
Length of output: 1324
🏁 Script executed:
#!/bin/bash
# Check if there are any safeguards or comments about fail-open lock behavior
rg -n "fail|open|redis.*unavailable|redis.*down" backend/app/core/distributed_lock.py -iRepository: wecode-ai/Wegent
Length of output: 938
🏁 Script executed:
#!/bin/bash
# Get full cleanup_worker function to see complete flow
sed -n '108,140p' backend/app/services/jobs.pyRepository: wecode-ai/Wegent
Length of output: 1258
Executor cleanup should fail-closed when the distributed lock cannot be reliably acquired.
acquire_watchdog_context_async() inherits fail-open behavior from acquire_async(): when Redis is unavailable or raises an exception, both return acquired=True. During a Redis outage, every pod can enter cleanup_stale_executors() concurrently, defeating the cross-pod coordination this lock is intended to provide. The fail-open strategy is documented in acquire_async() (line 117: "fail-open for availability"), but for deletion operations like executor cleanup, this creates a risk. Consider implementing a fail-closed lock mode for executor cleanup, or add an explicit Redis health check before executing deletion work.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/jobs.py` around lines 113 - 124, The current block uses
distributed_lock.acquire_watchdog_context_async (and thus acquire_async) which
is fail-open when Redis errors occur; change executor cleanup to fail-closed by
performing a Redis health check before running cleanup: before entering the
async with acquire_watchdog_context_async(EXECUTOR_CLEANUP_LOCK_KEY,...), call
the lock/redis health method (e.g., distributed_lock.ping() or a provided health
check on distributed_lock) and if the health check fails, log and skip cleanup;
alternatively, if acquire_watchdog_context_async supports a strict/fail_closed
flag, pass that option; ensure you still use EXECUTOR_CLEANUP_LOCK_KEY and only
call job_service.cleanup_stale_executors(db) when the health check passes and
the lock was truly acquired.
|
|
||
|
|
||
| def stop_background_jobs(app): | ||
| async def stop_background_jobs(app): |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Expect every runtime call to be either `await stop_background_jobs(...)`
# or registration as an async lifespan/shutdown callable that awaits it.
rg -nP -C3 '\bstop_background_jobs\s*\(' --glob '*.py'Repository: wecode-ai/Wegent
Length of output: 829
🏁 Script executed:
cat -n backend/app/main.py | sed -n '370,390p'Repository: wecode-ai/Wegent
Length of output: 914
🏁 Script executed:
# Search for the function containing the stop_background_jobs call
ast-grep --pattern 'function $_() { $$$ stop_background_jobs $$$ }'Repository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Check if there are any other variations of the call (with await, etc.)
rg -nP 'await\s+stop_background_jobs' --glob '*.py'Repository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Find the function signature that contains line 379
rg -n 'async def|def ' backend/app/main.py | head -20Repository: wecode-ai/Wegent
Length of output: 261
🏁 Script executed:
# Get more context around line 379 to find the function
sed -n '1,379p' backend/app/main.py | tail -50 | head -30Repository: wecode-ai/Wegent
Length of output: 1219
Add await to the stop_background_jobs(app) call in the lifespan shutdown phase.
The stop_background_jobs function is now async but is called without await at backend/app/main.py:379 within the async lifespan context manager. This will create an un-awaited coroutine, skipping the background job cleanup during shutdown.
Change stop_background_jobs(app) to await stop_background_jobs(app).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/jobs.py` at line 230, The shutdown call to the async
function stop_background_jobs(app) is not awaited inside the async lifespan
context manager, creating an un-awaited coroutine and skipping cleanup; update
the call site in the async lifespan shutdown phase to use await
stop_background_jobs(app) so the coroutine is executed and background jobs are
properly stopped.
Feat/executor cleanup cursor (wecode-ai#1014) See merge request weibo_rd/common/wecode/wegent!1996
Summary by CodeRabbit
New Features
Bug Fixes
Tests