Skip to content

Commit 6e5c3a9

Browse files
colbymchenryclaude
colbymchenry
and
claude
committed
feat(resolution): bridge Celery .delay()/.apply_async() dispatch to the task body
Celery decouples a task's call site from its body: a @shared_task / @app.task decorated def is invoked via task.delay(...) / task.apply_async(...), a dynamic hop with no static edge, so flows dead-end at the dispatch and the agent reads tasks.py to reconstruct them. celeryDispatchEdges links the enclosing function at each .delay/.apply_async site -> the task function body. Precision rests on a DECORATOR gate: the dispatched name must resolve to a Python function carrying a task decorator, read from the source lines ABOVE its def (the def's startLine excludes the decorator, and no decorates edge exists since @shared_task is an unresolved external import). The kind==='function' filter drops same-named test-method collisions; canvas forms (group(t).delay(), t.s()/.si()) have no single identifier before .delay so they're skipped, not mis-bridged; cross-module name collisions prefer a same-file task else bail. Surfaces as `dynamic: celery dispatch` via the generic synth-edge fallback. Validated 100% precision on two grep-confirmed repos exercising both decorator dialects: paperless-ngx (small, @shared_task, 31 edges, 31/31 real) and pretix (medium, @app.task, 63 edges across 21 tasks, 0/21 false positives); 0 on the httpie control (no Celery). Node-stable (pure edge synth). Suite 1615 green. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
1 parent 80a1044 commit 6e5c3a9

4 files changed

Lines changed: 233 additions & 2 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
1515
- `codegraph_explore` now connects a Vue component to the **Pinia** store action it calls. When code does `const store = useUserStore()` and then `store.fetchUser()`, that call now links through to the `fetchUser` action in the store module — so "what happens when this view loads its data?" traces from the component into the action's body instead of stopping at the `store.fetchUser()` line. Works for both Pinia store styles (options and setup), and stays precise (a built-in like `store.$patch()` or an unrelated same-named method isn't mislinked).
1616
- `codegraph_explore` now follows **Vuex** string dispatch. A `dispatch('user/login')` or `commit('SET_TOKEN')` call — namespaced `'module/action'` keys included — now links to the action or mutation it names, resolved to the correct store module even when several modules share an action name (and without being fooled by a same-named `api/` helper). So "what runs when this dispatches?" traces from the call into the store handler and on to the mutations it commits. Vuex's canonical `export default { namespaced, actions, mutations }` module shape is now indexed too, so those handlers are findable symbols.
1717
- `codegraph_explore` now connects React data-fetching flows built on **RTK Query** (Redux Toolkit's `createApi`). An endpoint defined inside `createApi({ endpoints })` and the `useGetXQuery` / `useUpdateYMutation` hook it generates were both invisible to analysis — so "what does this component fetch?" or "where does `useGetThingQuery` get its data?" dead-ended, because the hook, the endpoint, and the component had nothing linking them. CodeGraph now indexes each endpoint and each generated hook as real symbols and wires the path `component → useGetXQuery → getX → queryFn`, so the flow resolves in one explore call instead of reading the API slice by hand. Both the arrow (`endpoints: build => ({ … })`) and method (`endpoints(builder) { return { … } }`) styles are recognized, along with the `useLazyGetXQuery` variant; hand-written hooks of a similar name are left untouched.
18+
- `codegraph_explore` now follows **Celery** task dispatch in Python. A `send_email.delay(...)` or `send_email.apply_async(...)` call now links to the `@shared_task` / `@app.task` function it runs — typically defined in a different module (`tasks.py`) from where it's triggered (a view or service) — so "what actually happens when this is dispatched?" traces from the call site straight into the task body instead of stopping at the `.delay()` line. Both decorator dialects are recognized (bare `@shared_task` and the arg'd `@app.task(bind=True, …)` form), including the module-qualified `tasks.invalidate_cache.apply_async()` call style. It stays precise: a `.delay()` on something that isn't a Celery task is never mislinked, so a project that doesn't use Celery is unaffected.
1819

1920
- `codegraph_explore` now surfaces the right code in large multi-layer projects. When you ask a backend-flow question in a repo that pairs an API server with a big frontend that mirrors the same domain words — say an `app/` admin UI sitting over an `api/` server — the server-side file that genuinely matches several of your query's terms is no longer pushed out of the results by the larger, more interconnected frontend layer. A file corroborated by two or more distinct query terms is now kept in the answer even when a denser unrelated layer would otherwise crowd it out, so "how does X read items / handle the request" returns the service or handler that does the work instead of a wall of frontend views. Single-layer projects are unaffected; set `CODEGRAPH_RANK_NO_MULTITERM=1` to revert to the previous ranking.
2021
- Impact and blast-radius analysis for TypeScript, JavaScript, Go, Python, Rust, Ruby, C, Java, C#, PHP, Scala, Kotlin, Swift, Dart, and Pascal/Delphi now understands the readers of a constant. When you change a file-scope, package-level, module-level, or class-level constant — a config object, a lookup table, a shared constant — the other symbols in that file that read it now show up as affected, where before they were invisible (impact only followed calls, imports, and inheritance, so a constant's consumers looked like "nothing depends on this"). This makes `codegraph impact`, and the impact trail in `codegraph_explore`/`codegraph_node`, catch the "change this table, break its readers" class of change. It's on by default and adds no nodes to your graph; bundled/minified files and ambiguously-shadowed names are skipped to keep results precise. Set `CODEGRAPH_VALUE_REFS=0` to turn it off.

‎__tests__/celery-dispatch-synthesizer.test.ts‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
/**
2+
* Celery task-dispatch bridge (Python).
3+
*
4+
* Celery decouples a task's call site from its body: a `@shared_task` / `@app.task`
5+
* decorated `def` is invoked through `task.delay(...)` / `task.apply_async(...)`, a
6+
* dynamic hop with no static edge. This bridges each `.delay`/`.apply_async` site to
7+
* the task function, gated on the DECORATOR (read from the source above the `def`) so a
8+
* `.delay()` on a non-task object resolves to nothing. Covers both decorator dialects
9+
* (`@shared_task`, `@app.task(...)`), the module-qualified `mod.task.apply_async()` form,
10+
* and proves the precision gates: a plain function called with `.delay()` and a canvas
11+
* `group(...).delay()` (no single identifier before `.delay`) both contribute no edge.
12+
*/
13+
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
14+
import * as fs from 'node:fs';
15+
import * as path from 'node:path';
16+
import * as os from 'node:os';
17+
import { CodeGraph } from '../src';
18+
19+
describe('celery-dispatch synthesizer', () => {
20+
let dir: string;
21+
beforeEach(() => { dir = fs.mkdtempSync(path.join(os.tmpdir(), 'celery-dispatch-')); });
22+
afterEach(() => { fs.rmSync(dir, { recursive: true, force: true }); });
23+
24+
it('bridges .delay()/.apply_async() to decorated tasks, ignoring non-task and canvas dispatch', async () => {
25+
// Two decorator dialects: bare @shared_task and arg'd @app.task(...).
26+
fs.writeFileSync(
27+
path.join(dir, 'tasks.py'),
28+
`from celery import shared_task
29+
from myapp.celery import app
30+
31+
32+
@shared_task
33+
def send_email(to):
34+
return to
35+
36+
37+
@app.task(bind=True, max_retries=3)
38+
def crunch(self, n):
39+
return n * 2
40+
`
41+
);
42+
fs.mkdirSync(path.join(dir, 'services'), { recursive: true });
43+
fs.writeFileSync(
44+
path.join(dir, 'services', 'tickets.py'),
45+
`from celery import shared_task
46+
47+
48+
@shared_task
49+
def invalidate_cache():
50+
return None
51+
`
52+
);
53+
// A plain function — NOT a celery task — that nonetheless has .delay() called on it.
54+
fs.writeFileSync(
55+
path.join(dir, 'utils.py'),
56+
`def process_data(x):
57+
return x
58+
`
59+
);
60+
// Dispatch sites, all inside one enclosing function.
61+
fs.writeFileSync(
62+
path.join(dir, 'views.py'),
63+
`from tasks import send_email, crunch
64+
from services import tickets
65+
from utils import process_data
66+
from celery import group
67+
68+
69+
def handle_request(req):
70+
send_email.delay(req.addr) # → send_email task (cross-file)
71+
crunch.apply_async(args=[5]) # → crunch task (@app.task dialect)
72+
tickets.invalidate_cache.apply_async() # module-qualified → invalidate_cache
73+
process_data.delay(req.x) # NOT a task → no edge
74+
group([send_email.s(a) for a in req.addrs]).delay() # canvas → no edge
75+
`
76+
);
77+
78+
const cg = await CodeGraph.init(dir, { silent: true });
79+
await cg.indexAll();
80+
const db = (cg as any).db.db;
81+
82+
const edges = db
83+
.prepare(
84+
`SELECT s.name source, t.name target, t.file_path tf, json_extract(e.metadata,'$.via') via
85+
FROM edges e JOIN nodes s ON s.id = e.source JOIN nodes t ON t.id = e.target
86+
WHERE json_extract(e.metadata,'$.synthesizedBy') = 'celery-dispatch'`
87+
)
88+
.all();
89+
90+
const targets = (src: string) => edges.filter((r: any) => r.source === src).map((r: any) => r.target).sort();
91+
// handle_request dispatches exactly the three real tasks (both dialects + module-qualified).
92+
expect(targets('handle_request')).toEqual(['crunch', 'invalidate_cache', 'send_email']);
93+
// The @app.task target resolved to the task def, not anything else.
94+
const crunchEdge = edges.find((r: any) => r.target === 'crunch');
95+
expect(crunchEdge.tf).toMatch(/tasks\.py$/);
96+
// Module-qualified `tickets.invalidate_cache.apply_async()` resolved by the last identifier.
97+
const cacheEdge = edges.find((r: any) => r.target === 'invalidate_cache');
98+
expect(cacheEdge.tf).toMatch(/services[\\/]tickets\.py$/);
99+
expect(cacheEdge.via).toBe('invalidate_cache');
100+
// PRECISION: a plain function called with .delay() is never targeted (no decorator).
101+
expect(edges.some((r: any) => r.target === 'process_data')).toBe(false);
102+
103+
cg.close?.();
104+
});
105+
106+
it('produces no edges in a Celery-free project (clean control)', async () => {
107+
fs.writeFileSync(
108+
path.join(dir, 'app.py'),
109+
`def schedule(job):
110+
job.delay() # a .delay() that has nothing to do with Celery
111+
return job
112+
113+
114+
def run():
115+
schedule(make_job())
116+
`
117+
);
118+
const cg = await CodeGraph.init(dir, { silent: true });
119+
await cg.indexAll();
120+
const db = (cg as any).db.db;
121+
const count = db
122+
.prepare(
123+
`SELECT count(*) c FROM edges WHERE json_extract(metadata,'$.synthesizedBy') = 'celery-dispatch'`
124+
)
125+
.get();
126+
expect(count.c).toBe(0);
127+
cg.close?.();
128+
});
129+
});

‎docs/design/dispatch-synthesizer-backlog.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,8 @@ Status legend (matches the playbook): ✅ done+validated · 🟡 shipped but und
6464
| Shape | Ecosystem | Anchor | Mechanism | Status |
6565
|---|---|---|---|---|
6666
| **MediatR / CQRS** | .NET | `IRequest<T>` → `IRequestHandler<TReq,T>` by the generic request type; `_mediator.Send(new GetFooQuery())` → handler | S (generic-type-keyed) | 🔬 named a frontier in CLAUDE.md, but it's statically keyable via the generic — worth a real attempt |
67-
| **Celery / Sidekiq** | Python / Ruby | `@task`/`@shared_task` + `.delay()`/`.apply_async()`; `Worker.perform_async` → `perform` | R/X (decorator + name) | ⬜ |
67+
| **Celery** | Python | `@shared_task`/`@app.task`/`@<app>.task`/`@task` def + `.delay()`/`.apply_async()` call → task body | S (decorator-gated name) | ✅ **SHIPPED (2026-06-20)** — `synthesizedBy:'celery-dispatch'` (`celeryDispatchEdges`). Link the enclosing fn at each `.delay(`/`.apply_async(` site → the task fn. Precision rests on the DECORATOR gate: the dispatched name must resolve to a Python `function` carrying a task decorator, read from the source lines ABOVE its `def` (the def's own startLine excludes the decorator; no `decorates` edge exists — `@shared_task` is an unresolved external import). `kind==='function'` filter drops the same-named test-method collision (`consume_file`). Canvas forms (`group(t).delay()`, `t.s()`/`.si()`) have no single identifier before `.delay` → skipped, not mis-bridged. Cross-module name collision → same-file preference else bail. **100% precision: paperless-ngx (small, `@shared_task`, 31 edges, 31/31 real), pretix (medium, `@app.task`, 63 edges across 21 tasks, 0/21 FP); 0 on the httpie control (no Celery).** Node-stable (pure edge synth, no extraction change). Surfaces as `dynamic: celery dispatch @site` via the generic fallback. `+ celery-dispatch-synthesizer.test.ts`. **Deferred (recall):** canvas dispatch, class-based `Task` subclasses, `app.send_task('dotted.name')` string dispatch, aliased imports (`import send_email as s; s.delay()`). |
68+
| **Sidekiq** | Ruby | `class W; include Sidekiq::Job; def perform; end` + `W.perform_async(...)` → `perform` | S (class→perform) | ⬜ — the Ruby sibling of Celery; build next-to-it (grep-confirm ≥2 `perform_async` repos). |
6869
| **Laravel / Spring events** | PHP / Java | `event(OrderShipped::class)` → `EventServiceProvider` listener map; `@EventListener onX(EventT)` → publisher by event type | R (mapped) | ⬜ |
6970

7071
### Tier C — frontier, ⛔ do **not** build (no static anchor; would add noise)
@@ -82,6 +83,9 @@ Status legend (matches the playbook): ✅ done+validated · 🟡 shipped but und
8283
| Redux thunk | `redux-thunk` | ✅ **generalizes (2026-06-20)** — precise on uwave-web (small, 5 edges), session-desktop (medium, 2), trezor (large, 211); control shapeshift (RTK Query, no thunks) = 0. Receiver-agnostic (`api.dispatch`/`thunkApi.dispatch`/`window.…dispatch` all matched). **⚠️ 2 follow-ups below.** |
8384
| Object-literal registry | `object-registry` | ✅ **shipped (2026-06-20)** — xrengine `CommandManager` (64), Prebid.js (7), warp-drive (1); 0 false positives after 4 precision gates. |
8485
| RTK Query | `rtk-query` | ✅ **shipped (2026-06-20)** — 100% precision (hooks == synth edges, 0 cross-file) on basetool (54), minusx-metabase (11), shapeshift (13); 0 on uwave-web control. Extraction mints endpoint + generated-hook nodes; synth bridges hook→endpoint by convention. |
86+
| Pinia store | `pinia-store` | ✅ **shipped (2026-06-20)** — `useStore().action()` instance dispatch → action; 100% precision Geeker (41) / MallChat (64), 0 on element-admin (Vuex) control. |
87+
| Vuex dispatch | `vuex-dispatch` | ✅ **shipped (2026-06-20)** — string `dispatch('ns/action')`/`commit('M')` → handler; 100% precision element-admin (55) / vue-admin-template (12) / d2-admin (63), 0 on Redux controls. |
88+
| Celery | `celery-dispatch` | ✅ **shipped (2026-06-20)** — `.delay()`/`.apply_async()` → `@shared_task`/`@app.task` body; 100% precision paperless-ngx (31) / pretix (63 across 21 tasks), 0 on httpie control. Decorator-gated via source above the `def`. |
8589
| (see playbook §6 / `callback-synthesizer.ts` for the other ~20 channels) | | |
8690

8791
### redux-thunk follow-ups (found by the n>1 validation — this is exactly what it's for)

‎src/resolution/callback-synthesizer.ts‎

Lines changed: 98 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2092,12 +2092,107 @@ function vuexDispatchEdges(ctx: ResolutionContext): Edge[] {
20922092
return edges;
20932093
}
20942094

2095+
// ── Celery task dispatch (Python) ─────────────────────────────────────────────
2096+
// Celery decouples a task's call site from its body through async dispatch:
2097+
// # tasks.py
2098+
// @shared_task # also @app.task / @celery_app.task / @<app>.task / @task
2099+
// def process(account_ids): ...
2100+
// # views.py — a DIFFERENT module
2101+
// process.apply_async(kwargs={...}) # or process.delay(...) — dynamic, no static edge
2102+
// Bridge it: link the enclosing function/method at each `.delay(`/`.apply_async(` site → the
2103+
// task function body. Precision rests on the DECORATOR gate — the dispatched name must resolve
2104+
// to a Python function carrying a celery task decorator (read from the source lines above its
2105+
// `def`, since the def's own startLine excludes the decorator). A `.delay()` on a non-task
2106+
// object resolves to no task node → no edge, so a Celery-free repo yields 0. Same-file /
2107+
// unique-candidate disambiguation like vuex. (Canvas forms — `group(t).delay()`, `t.s()`/`.si()`
2108+
// — have no single identifier before `.delay`/`.apply_async`, so they're skipped, not mis-bridged.)
2109+
const CELERY_DISPATCH_RE = /\b([A-Za-z_]\w*)\s*\.\s*(?:delay|apply_async)\s*\(/g;
2110+
// A task decorator: bare `@shared_task`/`@task` or attribute `@app.task`/`@celery_app.task`,
2111+
// each optionally called with args. `\b`-bounded and `@`-anchored so `@mytask`, or a symbol
2112+
// merely named `task`, can't match. No `/g`, so `.test()` is stateless across reuse.
2113+
const CELERY_TASK_DECORATOR_RE = /@\s*(?:[A-Za-z_][\w.]*\.)?(?:shared_task|task)\b/;
2114+
const CELERY_PY_EXT = /\.py$/;
2115+
const CELERY_FANOUT_CAP = 80;
2116+
const CELERY_DECORATOR_LOOKBACK = 12; // max lines above a `def` to scan for its decorators
2117+
2118+
function celeryDispatchEdges(ctx: ResolutionContext): Edge[] {
2119+
// Memoize the decorator check per task-candidate node: it reads the file and scans a few
2120+
// lines above the def. Only called on names that are actually `.delay`/`.apply_async`
2121+
// receivers, so the candidate set stays small.
2122+
const taskCache = new Map<string, boolean>();
2123+
const isCeleryTask = (node: Node): boolean => {
2124+
let v = taskCache.get(node.id);
2125+
if (v !== undefined) return v;
2126+
v = false;
2127+
if (node.kind === 'function' && CELERY_PY_EXT.test(node.filePath)) {
2128+
const content = ctx.readFile(node.filePath);
2129+
if (content) {
2130+
const lines = content.split('\n');
2131+
// startLine is the `def` line (decorators sit ABOVE it). Walk upward, stopping at the
2132+
// previous declaration so a non-task def can never inherit the prior def's decorator.
2133+
const stop = Math.max(0, node.startLine - 1 - CELERY_DECORATOR_LOOKBACK);
2134+
for (let i = node.startLine - 2; i >= stop; i--) {
2135+
const t = (lines[i] ?? '').trim();
2136+
if (/^(?:async\s+def|def|class)\b/.test(t)) break; // previous decl → stop
2137+
if (CELERY_TASK_DECORATOR_RE.test(t)) { v = true; break; }
2138+
}
2139+
}
2140+
}
2141+
taskCache.set(node.id, v);
2142+
return v;
2143+
};
2144+
2145+
const resolve = (name: string, dispatchFile: string): Node | null => {
2146+
const cands = ctx.getNodesByName(name).filter((n) => n.kind === 'function' && isCeleryTask(n));
2147+
if (!cands.length) return null;
2148+
if (cands.length === 1) return cands[0]!;
2149+
// Cross-module name collision: prefer a task defined in the dispatching file, else bail
2150+
// (ambiguous — precision over recall, like vuex's root-key resolution).
2151+
return cands.find((c) => c.filePath === dispatchFile) ?? null;
2152+
};
2153+
2154+
const edges: Edge[] = [];
2155+
const seen = new Set<string>();
2156+
for (const file of ctx.getAllFiles()) {
2157+
if (!CELERY_PY_EXT.test(file)) continue;
2158+
const content = ctx.readFile(file);
2159+
if (!content || (!content.includes('.delay(') && !content.includes('.apply_async('))) continue;
2160+
const safe = stripCommentsForRegex(content, 'python');
2161+
const nodesInFile = ctx.getNodesInFile(file);
2162+
CELERY_DISPATCH_RE.lastIndex = 0;
2163+
let m: RegExpExecArray | null;
2164+
let added = 0;
2165+
while ((m = CELERY_DISPATCH_RE.exec(safe)) && added < CELERY_FANOUT_CAP) {
2166+
const name = m[1]!;
2167+
const line = safe.slice(0, m.index).split('\n').length;
2168+
const disp = enclosingFn(nodesInFile, line);
2169+
if (!disp) continue; // module-level dispatch — no source symbol to attribute
2170+
const target = resolve(name, file);
2171+
if (!target || target.id === disp.id) continue;
2172+
const key = `${disp.id}>${target.id}`;
2173+
if (seen.has(key)) continue;
2174+
seen.add(key);
2175+
edges.push({
2176+
source: disp.id,
2177+
target: target.id,
2178+
kind: 'calls',
2179+
line,
2180+
provenance: 'heuristic',
2181+
metadata: { synthesizedBy: 'celery-dispatch', via: name, registeredAt: `${file}:${line}` },
2182+
});
2183+
added++;
2184+
}
2185+
}
2186+
return edges;
2187+
}
2188+
20952189
/**
20962190
* Synthesize dispatcher→callback edges (field observers + EventEmitters +
20972191
* React re-render + JSX children + Vue templates + SvelteKit load + RN event
20982192
* channel + Fabric native-impl + MyBatis Java↔XML + Gin middleware chain +
20992193
* Redux-thunk dispatch chain + object-literal registry dispatch + RTK Query
2100-
* generated-hook → endpoint + Pinia useStore().action() + Vuex string dispatch).
2194+
* generated-hook → endpoint + Pinia useStore().action() + Vuex string dispatch +
2195+
* Celery task .delay()/.apply_async() → task body).
21012196
* Returns the count added. Never throws into indexing — callers wrap in try/catch.
21022197
*/
21032198
export function synthesizeCallbackEdges(queries: QueryBuilder, ctx: ResolutionContext): number {
@@ -2140,6 +2235,7 @@ export function synthesizeCallbackEdges(queries: QueryBuilder, ctx: ResolutionCo
21402235
const rtkEdges = rtkQueryEdges(queries, ctx);
21412236
const piniaEdges = piniaStoreEdges(ctx);
21422237
const vuexEdges = vuexDispatchEdges(ctx);
2238+
const celeryEdges = celeryDispatchEdges(ctx);
21432239

21442240
const merged: Edge[] = [];
21452241
const seen = new Set<string>();
@@ -2168,6 +2264,7 @@ export function synthesizeCallbackEdges(queries: QueryBuilder, ctx: ResolutionCo
21682264
...rtkEdges,
21692265
...piniaEdges,
21702266
...vuexEdges,
2267+
...celeryEdges,
21712268
]) {
21722269
const key = `${e.source}>${e.target}`;
21732270
if (seen.has(key)) continue;

0 commit comments

Comments
 (0)