What you'll build: sales-pipeline forecasting on one DataK3 bucket — the thing a sales leader pays a Clari or Salesforce Forecasting seat for. Your stage model is data (pipeline_stages rows, each with a probability and a forecast category); the weighted pipeline is Σ(amount × probability) in one SQL statement; forecast categories (commit / best-case / pipeline) roll up into periodic forecast_snapshots; and — the read no flat query gives you — the account graph rolls the weighted number up by corporate family. An optional AI risk gate (kimi-k2.6) catches the deal that's been sitting in commit untouched for six weeks and downgrades it, with a reason. The deterministic rollups are pure SQL; the model is the override, not the baseline.

This skill consumes the opportunities and accounts masters that crm/core owns (Build a CRM on DataK3); it owns the stage model, the snapshots, and the risk verdicts. Standalone, it stubs the two masters so you can run it end to end.

What you'll learn:

  • Model your sales process as pipeline_stages rows — a probability and a forecast category per stage.
  • Extend opportunities in place (forecast_category, probability) — an ALTER-add, no second copy.
  • Compute the weighted pipeline and roll it up by forecast category into forecast_snapshots.
  • Roll the weighted forecast up by corporate family with a graph traversal JOINed to opportunities.
  • Gate stale deals through kimi-k2.6: override the stage's default category, write deal_risk.
  • Deploy a pure-SQL crm-forecaster engine whose service account needs only k3.editor.

The problem — and why it matters

A sales leader's forecast lives in a spreadsheet that's stale the moment it's pasted. The paid answer — a weighted pipeline, commit / best-case / pipeline categories, and a per-rep, per-quarter roll — is a $30k+/yr forecasting seat bolted onto the CRM, syncing overnight. Two things it still gets wrong: it can't roll up a corporate family (Acme Corp + Acme Labs + Acme EU as one number) without a second graph database, and its "commit" bucket trusts the rep — the deal nobody has touched in six weeks still counts.

On DataK3 the whole thing is one bucket: three merge-keyed tables, a graph JOIN over the same rows, and one Models call. The stage model is data you edit, not a config screen. The family rollup is a traversal of the account hierarchy you already have. And the risk gate is the cheap part — it only runs on open deals, and it only ever overrides a deterministic baseline with a written reason.

PieceLands inPillar / runs on
Stage model (probability, forecast category)table pipeline_stagesSQL
Weighted pipeline Σ amount×probabilityquery over opportunities ⋈ pipeline_stagesSQL
Periodic forecast by categorytable forecast_snapshotsSQL
Family-weighted rollupcrm_graph (graph_khop) JOINed to opportunitiesGraph
Deal-risk / category overridetable deal_riskIgnite Models (kimi-k2.6)
The recompute enginewrites forecast_snapshotsIgnite app (pure SQL, scale-to-zero)

NOTE

Connect the DODIL MCP once — see the two-minute setup. Every step shows an Ask your agent tab (the default — DODIL is agent-native) and the CLI.

Prerequisites

  • The dodil CLI (dodil auth login) or the DODIL MCP connected to your agent (Claude Code, Cursor, VS Code, Codex).
  • export BUCKET=crm — one bucket is the whole system's data plane.
  • The crm/core masters opportunities and accounts (with the crm_graph account hierarchy). If you don't have them yet, Step 0 stubs the minimum this skill reads.

Step 0 — Stub the masters you consume (skip if you have crm/core)

Forecasting reads two masters crm/core owns: opportunities (the deals) and accounts (the hierarchy behind the family rollup). If your bucket already has them, skip to Step 1. Standalone, create them and seed the anchor dataset — the Acme corporate family plus independent Greyparrot — so every assertion below has real rows to hit.

You

In bucket crm, create merge-keyed masters opportunities (key opportunity_id: name, account_domain, primary_contact_email, pipeline, stage, status, amount(double), owner, source, close_date, created_at) and accounts (key org_domain: name, parent_domain, tier, country, industry, employee_band, annual_revenue(double)). Then upsert 4 accounts — Acme Corp (acme.io, parent none), Acme Labs (labs.acme.io, parent acme.io), Acme EU (acme.eu, parent acme.io), Greyparrot (greyparrot.ai, parent none) — and 4 open opportunities: opp-1 Acme Labs negotiate $24000, opp-2 Acme EU proposal $12000, opp-3 Greyparrot demo $30000, opp-4 Greyparrot qualify $8000, all owner [email protected], pipeline sales.

DODIL MCP tools called
data_bucket_create→data_table_create→data_table_upsert
Agent

Created bucket crm with opportunities (key opportunity_id) and accounts (key org_domain), then upserted 4 accounts (Acme family + Greyparrot) and 4 open opportunities keyed opp-1..opp-4. parent_domain='none' marks a top-level account (never empty-string — that drops the key).

To wire the family rollup (Step 4), project the accounts into crm/core's account graph — an integer-keyed crm_node table, a crm_edge table tagged by rel, and CREATE GRAPH. Populate both tables fully, then CREATE GRAPH — the graph snapshots its edges at creation, so edges added later aren't traversable until you re-create it. (If you ran crm/core with hierarchy=true, this already exists — skip it.)

You

In crm, build the account graph crm_graph: crm_node (id bigint KEY, kind, biz_key, name) with the 4 accounts as ids 1-4, and crm_edge (src, dst, rel) with subsidiary_of edges Labs->Corp and EU->Corp. Then CREATE GRAPH crm_graph over them.

DODIL MCP tools called
data_pg
Agent

crm_node has 4 account rows (ids 1-4); crm_edge has 2 subsidiary_of edges (Labs->Corp, EU->Corp). CREATE GRAPH crm_graph NODES (crm_node KEY id) EDGES (crm_edge SRC src DST dst) succeeded — edges snapshotted at creation.

Step 1 — Model your sales process as data (pipeline_stages)

The forecast is only as good as the stage model behind it, so make the model data you edit, not code you ship. pipeline_stages is one row per stage carrying two decisions: the probability that drives the weighted number, and the forecast category (pipeline | best_case | commit | omitted | closed) that buckets it. is_won / is_closed mark the terminal stages. This is the stages + probabilities params, seeded as parallel lists.

You

In crm, create pipeline_stages (key stage_id: pipeline, name, position(int), probability(double), forecast_category, is_won(boolean), is_closed(boolean)) and seed the sales pipeline: prospect 0.1 pipeline, qualify 0.25 pipeline, demo 0.4 best_case, proposal 0.6 best_case, negotiate 0.8 commit, closed_won 1.0 closed (is_won,is_closed), closed_lost 0.0 omitted (is_closed).

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created pipeline_stages (key stage_id) and seeded 7 stages for the 'sales' pipeline. Each stage carries its probability (0.1→1.0) and forecast_category; closed_won is is_won+is_closed, closed_lost is is_closed. Editing your process is now an upsert, not a redeploy.

Now extend opportunities in place — add forecast_category and probability so every deal carries its own (overridable) forecast fields, and materialize them from the stage each deal is in. This is an ALTER TABLE … ADD COLUMN — a column-add on the master, never a second copy of the deals.

You

In crm, add two columns to opportunities: forecast_category (varchar) and probability (double). Then set each open opportunity's probability and forecast_category from the pipeline_stages row matching its pipeline+stage.

DODIL MCP tools called
data_pg
Agent

Added opportunities.forecast_category and opportunities.probability. Backfilled all 4 open deals from pipeline_stages: opp-1 negotiate→commit/0.8, opp-2 proposal→best_case/0.6, opp-3 demo→best_case/0.4, opp-4 qualify→pipeline/0.25.

Step 2 — The weighted pipeline (Σ amount × probability)

Here's the number the spreadsheet fakes. The weighted pipeline is every open deal's amount times its stage probability, summed — computed live over one copy of the rows. Because Step 1 materialized probability onto each deal, you get the same answer whether you read the deal's column or JOIN back to pipeline_stages (that's the harmonization: the deal's forecast fields are a snapshot of the stage).

You

In crm, show the total open pipeline and the weighted pipeline (sum of amount×probability) over all open opportunities — computed both from the deal's own probability column and by joining to pipeline_stages, to prove they agree.

DODIL MCP tools called
data_sql
Agent

Open pipeline $74,000 across 4 deals; weighted pipeline $40,400 — identical from opportunities.probability and from the pipeline_stages JOIN (24000×0.8 + 12000×0.6 + 30000×0.4 + 8000×0.25).

Step 3 — Roll the forecast up by category (forecast_snapshots)

A forecast isn't one number — it's the commit (deals you'll call), the best-case (upside), and the raw pipeline, per rep, per period. Group the open book by (period, pipeline, owner, forecast_category) and write one forecast_snapshots row per bucket: open_amount, weighted_amount = Σ amount×probability, and committed_amount (the commit bucket only). The snapshot id is period:pipeline:owner:category, so re-running the roll upserts the same rows — a period's forecast is idempotent. With forecast_periods=quarterly, the period key is 2026-Q3.

You

In crm, create forecast_snapshots (key snapshot_id: period, pipeline, owner, forecast_category, open_amount(double), weighted_amount(double), committed_amount(double), deal_count(int), snapshot_at). Then recompute the 2026-Q3 sales forecast: group open opportunities by owner and forecast_category, and upsert one row per bucket with open_amount, weighted_amount=sum(amount×probability), committed_amount (commit only), and deal_count. Key each row period:pipeline:owner:category so re-runs upsert.

DODIL MCP tools called
data_table_create→data_sql→data_table_upsert
Agent

Created forecast_snapshots. Recomputed 2026-Q3/sales/[email protected] into 3 category rows — commit: 1 deal, open $24,000, weighted $19,200, committed $24,000; best_case: 2 deals, open $42,000, weighted $19,200; pipeline: 1 deal, open $8,000, weighted $2,000. Re-running upserts the same 3 rows.

Step 4 — Roll the forecast up by corporate family (the graph read)

This is the read a flat CRM can't give you. Acme Corp owns Acme Labs and Acme EU; a VP wants the family's weighted forecast as one number. Every subsidiary points up at Acme Corp (graph node 1), so a reverse k-hop from node 1 collects the whole family, and you JOIN that straight to opportunities and pipeline_stages to weight it — in one statement. Independent Greyparrot and its $30k of deals are correctly excluded: they're not in the tree.

You

In crm, roll up the weighted pipeline for Acme Corp's whole corporate family: reverse-traverse crm_graph from node 1 up to 2 hops to get the family accounts, JOIN to open opportunities and pipeline_stages, and sum amount and amount×probability. Greyparrot should be excluded.

DODIL MCP tools called
data_pg
Agent

Acme family (Acme Labs + Acme EU) = 2 open deals, open pipeline $36,000, weighted $26,400 (24000×0.8 + 12000×0.6). Greyparrot's $30k demo is excluded — it's not under Acme Corp in the graph.

NOTE

graph_khop('crm_graph', 1, 2, 'in') walks incoming edges, so it climbs down the ownership tree from the parent. The literal start key (1) lets the traversal JOIN straight to opportunities and pipeline_stages — graph and SQL over one copy of the rows. If crm/core was built with hierarchy=false (no graph), skip this step; the per-owner rollup in Step 3 still stands.

This untyped walk is only safe while the graph carries nothing but hierarchy edges — as here, in the standalone build. The moment other components project more relationship kinds into the shared graph (account-360 adds partner_of / supplies / competes_with), any incoming edge counts as "family" — which is exactly the bug the suite hit live. The package's family route uses a typed traversal for that reason; the story is in ## Routes.

Step 5 — The risk gate: override stale commits (kimi-k2.6)

The deterministic stage→category mapping is the baseline. But a deal parked in negotiate (commit) that nobody has touched in 45 days shouldn't count as commit — and that's a judgement call, so it's the one place a model earns its keep. The gate takes an opportunity's stage, amount, days_in_stage, last_activity_days, and the stage's default category, and returns only JSON: a possibly-downgraded forecast_category, a risk (on_track | slipping | at_risk), and a short reason. It runs only on open deals, and it only ever overrides the baseline — so it's cheap and auditable. The verdict lands in deal_risk.

You

Assess forecast risk for opp-1 on kimi-k2.6: stage negotiate, default_forecast_category commit, amount 24000, days_in_stage 45, last_activity_days 45. Return ONLY JSON {forecast_category, risk (on_track|slipping|at_risk), reason (<=20 words)} — downgrade a stale commit.

DODIL MCP tools called
ignite_models_chat
Agent

{"forecast_category": "best_case", "risk": "at_risk", "reason": "Untouched for 45 days in negotiation; stale commit downgraded."}

Create deal_risk and write the verdict, keyed on opportunity_id so a re-assessment upserts:

You

In crm, create deal_risk (key opportunity_id: forecast_category, risk, days_in_stage(int), last_activity_days(int), reason, assessed_at) and upsert the opp-1 verdict — forecast_category best_case, risk at_risk, 45 days in stage, 45 since last activity, reason 'Untouched for 45 days in negotiation; stale commit downgraded.'

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created deal_risk (key opportunity_id) and upserted opp-1: commit→best_case, at_risk, days_in_stage 45. A forecast that reads deal_risk over the stage default now discounts this deal out of commit.

Step 6 — Deploy the recompute engine (crm-forecaster, pure SQL)

The snapshot roll (Step 3) should run on a schedule, not by hand — so it's a small Ignite app, deployed in image mode: the platform Kaniko-builds a Dockerfile on deploy and runs the container, so the handler is a tiny HTTP server (GET /healthz, POST /recompute) — not a handler(payload, ctx) function. It's a separate workload, so it gets its own service account to write DataK3 (client-credentials → bearer token, injected as runtime env). The key point: this handler is pure SQL — it never calls Models — so its service account needs only k3.editor. (The risk gate in Step 5 is a separate, optional concern; enable it in-handler and you'd add ignite.model-user. At defaults the forecaster stays SQL-only.)

Data goes in and out over the drop-in Postgres wire (pg.uk-lon-1.dodil.io:5432, dbname = bucket, user=token, password = the SA access token) via psycopg — there is no K3 HTTP API. Re-running the forecast re-writes the same snapshot_id, so the write is INSERT … ON CONFLICT (snapshot_id) DO UPDATE (a bare INSERT of an already-committed PK raises duplicate-key 23505 — a plain INSERT is not an upsert on re-write; DuckDB pg-wire supports ON CONFLICT, or use the managed data_table_upsert); retry on a serialization failure.

# forecaster/server.py — IMAGE-mode Ignite app: an HTTP server that recomputes
# forecast_snapshots for one period. PURE SQL — it never calls Models.
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
BUCKET    = os.environ["BUCKET"]
SA_ID     = os.environ["DODIL_SERVICE_ACCOUNT_ID"]      # the cli-… serviceAccountId, NOT the uuid
SA_SECRET = os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]
PERIODS   = os.environ.get("FORECAST_PERIODS", "quarterly")   # quarterly | monthly
PG_HOST   = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT   = int(os.environ.get("PG_PORT", "5432"))
 
ID_URL = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
# an explicit User-Agent is REQUIRED — stdlib urllib's default "Python-urllib/x" is
# banned by Cloudflare at id.dodil.io (HTTP 403 "error code: 1010").
UA = "crm-forecaster/1.0"
 
FS_COLS = ["snapshot_id", "period", "pipeline", "owner", "forecast_category",
           "open_amount", "weighted_amount", "committed_amount", "deal_count", "snapshot_at"]
 
ROLL = """
  SELECT o.owner AS owner, ps.forecast_category AS forecast_category,
         count(*)                                                                     AS deal_count,
         round(sum(o.amount),2)                                                       AS open_amount,
         round(sum(o.amount*ps.probability),2)                                        AS weighted_amount,
         round(sum(CASE WHEN ps.forecast_category='commit' THEN o.amount ELSE 0 END),2) AS committed_amount
  FROM opportunities o
  JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage
  WHERE o.status='open' AND o.pipeline=%s
  GROUP BY o.owner, ps.forecast_category"""
 
 
def _now():
    return datetime.now(timezone.utc).isoformat()
 
 
def _http_post(url, data, headers, form=False):
    headers = {"User-Agent": UA, **headers}
    if form:
        body = urllib.parse.urlencode(data).encode()
        headers["Content-Type"] = "application/x-www-form-urlencoded"
    else:
        body = json.dumps(data).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=90) as r:
        return json.loads(r.read().decode())
 
 
def _token():
    out = _http_post(ID_URL, {"grant_type": "client_credentials",
                              "client_id": SA_ID, "client_secret": SA_SECRET},
                     headers={}, form=True)
    return out["access_token"]
 
 
def _pg(token):
    # drop-in Postgres wire: DB name = bucket, user "token", password = the SA access token.
    return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET,
                           user="token", password=token, sslmode="require",
                           connect_timeout=20, autocommit=False)
 
 
def _retry(fn):
    # the pg engine is serializable — retry a write on a transient serialization/deadlock.
    for attempt in range(4):
        try:
            return fn()
        except (pg_errors.SerializationFailure, pg_errors.DeadlockDetected):
            if attempt == 3:
                raise
            time.sleep(0.4 * (attempt + 1))
 
 
def _current_period():
    d = datetime.now(timezone.utc)
    if PERIODS == "monthly":
        return f"{d.year}-{d.month:02d}"                  # 2026-09
    return f"{d.year}-Q{(d.month - 1) // 3 + 1}"          # 2026-Q3
 
 
def _write_snapshot(token, row):
    # snapshot_id is deterministic (period:pipeline:owner:category — no timestamp), so re-running the
    # forecast for the same period re-writes the SAME PK. This MUST be an ON CONFLICT upsert: a bare
    # re-INSERT of a committed PK raises duplicate-key 23505. DuckDB pg-wire supports ON CONFLICT
    # (verified live); the managed data_table_upsert is the alternative.
    setc = ", ".join(f"{c} = EXCLUDED.{c}" for c in FS_COLS if c != "snapshot_id")
    insert = (f"INSERT INTO forecast_snapshots ({', '.join(FS_COLS)}) "
              f"VALUES ({', '.join(['%s'] * len(FS_COLS))}) "
              f"ON CONFLICT (snapshot_id) DO UPDATE SET {setc}")
    vals = [row[c] for c in FS_COLS]
    def _w():
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute(insert, vals)
            conn.commit()
    _retry(_w)
 
 
def recompute(period, pipeline):
    token = _token()
    with _pg(token) as conn, conn.cursor() as cur:       # the pure-SQL roll — no Models
        cur.execute(ROLL, (pipeline,))
        cols = [c.name for c in cur.description]
        buckets = [dict(zip(cols, r)) for r in cur.fetchall()]
    for b in buckets:                                    # one snapshot row per (owner, category)
        _write_snapshot(token, {
            "snapshot_id": f"{period}:{pipeline}:{b['owner']}:{b['forecast_category']}",
            "period": period, "pipeline": pipeline, "owner": b["owner"],
            "forecast_category": b["forecast_category"], "open_amount": b["open_amount"],
            "weighted_amount": b["weighted_amount"], "committed_amount": b["committed_amount"],
            "deal_count": b["deal_count"], "snapshot_at": _now()})
    return {"period": period, "pipeline": pipeline, "snapshots_written": len(buckets)}
 
 
class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)
 
    def do_GET(self):
        if self.path == "/healthz":
            return self._send(200, {"status": "ready"})     # probe path — no auth
        return self._send(404, {"error": "no_route", "path": self.path})
 
    def do_POST(self):
        if self.path != "/recompute":
            return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            body = json.loads(self.rfile.read(n) or b"{}")
            period   = body.get("period") or _current_period()
            pipeline = body.get("pipeline", "sales")
            return self._send(200, recompute(period, pipeline))
        except Exception as e:
            return self._send(500, {"error": type(e).__name__, "detail": str(e)[:600]})
 
    def log_message(self, *a):
        pass
 
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    print(f"crm-forecaster serving on 0.0.0.0:{port} bucket={BUCKET} periods={PERIODS}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

Two more files sit next to it. The Dockerfile is what the platform builds on deploy; requirements.txt pins the one third-party dep (the pg driver — everything else is stdlib):

# forecaster/Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
ENV PORT=8080
EXPOSE 8080
CMD ["python", "server.py"]
# forecaster/requirements.txt
psycopg[binary]==3.2.3

Give it a least-privilege identity — k3.editor and nothing else — then build-and-deploy the image and smoke-test it with a POST:

You

Create a service account crm-forecaster-sa, grant it ONLY k3.editor (the handler is pure SQL — no Models), then image-mode deploy my ./forecaster app to Ignite as crm-forecaster (Dockerfile build-on-deploy, port 8080, health /healthz) with the service-account creds and BUCKET as runtime env, and POST to /recompute for the 2026-Q3 sales period as a smoke test.

DODIL MCP tools called
auth_service-account_create→auth_service-account_grant-role→ignite_app_deploy
Agent

Created crm-forecaster-sa, granted k3-authorization-service k3.editor (no ignite.model-user — pure SQL). Image-deployed crm-forecaster (Dockerfile Kaniko build-on-deploy; scale-to-zero) at crm-forecaster-$ORG-8080.ignite.dodil.cloud. POST /recompute {period:2026-Q3} returned {snapshots_written: 3} — commit/best_case/pipeline rewritten for the period.

Recurring recompute has no server-side scheduler — POST /recompute from your own scheduler (cron → curl), or pin the app warm with --auto-min-instances 1 to avoid a cold-start 502 on the first hit. Scale-to-zero (the default) is right for a nightly forecast roll.

Query it — one bucket, three pillars, drop-in clients

Now it's a forecast you query — the weighted book by category (SQL), the family roll (graph), the at-risk deals (the gate's output) — over one bucket. Any Postgres client points straight at it: data connect prints the endpoints (DB name = bucket, credential = your login token).

You

In crm, show the 2026-Q3 forecast by category from forecast_snapshots, list the at-risk deals from deal_risk, and print the endpoint to point psql at the crm bucket.

DODIL MCP tools called
data_sql→data_connect
Agent

2026-Q3: commit $24,000 (weighted $19,200), best_case $42,000 (weighted $19,200), pipeline $8,000 (weighted $2,000). At risk: opp-1 (commit→best_case, at_risk). data connect printed pg pg.uk-lon-1.dodil.io:5432/crm — drop-in psql.

Routes

The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py — CRUD over the stage model + masters, plus the four questions a flat CRM can't answer: a weighted pipeline, an idempotent forecast recompute, a corporate-family rollup, and the AI risk gate. This is what the crm-forecaster engine (Step 6) wraps. The routes live on an APIRouter — the suite app mounts all seven CRM components on one FastAPI under per-component prefixes (this one at /pipeline-forecast) — while app = FastAPI(...) at the bottom keeps the package independently runnable (uvicorn routes:app). Every route follows the same three DataK3 rules the package bakes in, so quoting it is documenting them.

The connection and the one write helper live in db.py. A DataK3 bucket is a Postgres endpoint — db name = the bucket, user = the literal token, password = your DODIL token — so there's no data connect step in code, just fixed region constants. upsert() is the only writer every route uses — INSERT … ON CONFLICT DO UPDATE, so a retry, a shard replay, or a re-import lands the row once (a bare re-INSERT of a committed PK raises duplicate-key 23505). Money is Numeric(18, 2) end to end — DECIMAL over the pg wire, never coerced through the gRPC path that can drop an integer 0 into a DECIMAL column.

CRUD — each write is an upsert then a commit (DataK3 has no read-your-writes inside an open transaction; the engine is expire_on_commit=False, so routes commit before they return):

# routes.py — CRUD over the stage model + masters, keyed on the natural PK
@router.post("/pipeline_stages")
def upsert_stage(p: PipelineStageIn, s: Session = Depends(db)):
    upsert(s, PipelineStage, [p.model_dump()], key="stage_id")
    s.commit()
    return {"ok": True, "stage_id": p.stage_id}
 
 
@router.post("/opportunities")
def upsert_opportunity(o: OpportunityIn, s: Session = Depends(db)):
    upsert(s, Opportunity, [o.model_dump()], key="opportunity_id")
    s.commit()
    return {"ok": True, "opportunity_id": o.opportunity_id}

Workflow op 1 — the weighted pipeline (SQL). GET /pipeline/weighted sums amount × probability over the open book, computed both from the deal's own probability column and by a JOIN back to pipeline_stages, to prove they agree. Re-validated 2026-09-06 on the persistent crm bucket over the suite's opportunity book: open 63000, weighted 20100 from both paths, 4 deals (after POST /pipeline/backfill has materialized each deal's stage fields):

# routes.py — workflow op 2: the weighted pipeline (Σ amount × probability)
@router.get("/pipeline/weighted")
def weighted_pipeline(pipeline: str = "sales", s: Session = Depends(db)):
    row = s.execute(text(
        "SELECT round(sum(o.amount), 2)                AS open_amount, "
        "       round(sum(o.amount*ps.probability), 2) AS weighted_via_stage, "
        "       round(sum(o.amount*o.probability), 2)  AS weighted_via_col, "
        "       count(*)                               AS deal_count "
        "FROM opportunities o "
        "JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage "
        "WHERE o.status='open' AND o.pipeline=:p"
    ), {"p": pipeline}).mappings().one()
    return {k: (float(v) if hasattr(v, "__float__") else v) for k, v in row.items()}

Workflow op 2 — the recompute (SQL, idempotent). POST /forecast/recompute groups the open book by (owner, forecast_category) and upserts one forecast_snapshots row per bucket. Because snapshot_id is period:pipeline:owner:category (no timestamp), re-running the same period re-writes the same rows — this is the whole point of a recompute engine. Live (re-validated 2026-09-06): the first run writes the period's bucket rows; a second run returns the same count and the table holds exactly those rows, not double — it never double-counts:

# routes.py — workflow op 3: recompute the period's forecast (idempotent)
@router.post("/forecast/recompute")
def recompute(r: RecomputeIn, s: Session = Depends(db)):
    buckets = s.execute(text(
        "SELECT o.owner AS owner, ps.forecast_category AS forecast_category, "
        "       count(*)                                                                      AS deal_count, "
        "       round(sum(o.amount), 2)                                                       AS open_amount, "
        "       round(sum(o.amount*ps.probability), 2)                                        AS weighted_amount, "
        "       round(sum(CASE WHEN ps.forecast_category='commit' THEN o.amount ELSE 0 END), 2) AS committed_amount "
        "FROM opportunities o "
        "JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage "
        "WHERE o.status='open' AND o.pipeline=:p "
        "GROUP BY o.owner, ps.forecast_category"
    ), {"p": r.pipeline}).mappings().all()
    rows = [{
        "snapshot_id": f"{r.period}:{r.pipeline}:{b['owner']}:{b['forecast_category']}",
        "period": r.period, "pipeline": r.pipeline, "owner": b["owner"],
        "forecast_category": b["forecast_category"], "open_amount": b["open_amount"],
        "weighted_amount": b["weighted_amount"], "committed_amount": b["committed_amount"],
        "deal_count": b["deal_count"], "snapshot_at": _now(),
    } for b in buckets]
    if rows:
        upsert(s, ForecastSnapshot, rows, key="snapshot_id")
        s.commit()
    return {"period": r.period, "pipeline": r.pipeline, "snapshots_written": len(rows)}

Workflow op 3 — the corporate-family rollup (GRAPH), and the bug that rewrote it. The first version of GET /forecast/family/{node_id} walked the graph untyped — cypher('crm_graph', 'MATCH (root)<-[*1..5]-(child) …'), any incoming edge counts — and it was correct for as long as this component was validated on its own bucket, where crm_graph held nothing but subsidiary_of edges. The first time the whole suite ran on one bucket, account-360's multi-relational edges landed in the same shared graph — and the family walk quietly pulled Globex, a supplier of Acme, into Acme's corporate family, inflating the family forecast by $18k. No error, no warning: a supplier's supplies edge points at Acme exactly like a subsidiary's subsidiary_of edge does, and the embedded cypher() subset supports only an id(<var>) = <key> anchor — it cannot filter on an edge's rel. That's the composition lesson: a traversal that's correct on an isolated graph can be wrong on a shared one, and only integration testing on the shared bucket surfaces it. The fix is a typed recursive CTE over crm_edge with an explicit rel = 'subsidiary_of' filter (a top-level SELECT, integer-literal root, node ids fed into a SQL IN (…) — DuckDB has no = ANY(array)):

# routes.py — workflow op 4: roll the forecast up by corporate family (GRAPH)
@router.get("/forecast/family/{node_id}")
def family_forecast(node_id: int, s: Session = Depends(db)):
    """Roll the weighted pipeline up over a whole corporate family — following ONLY
    `subsidiary_of` edges.
 
    TYPED traversal, deliberately: the suite's shared `crm_graph` is multi-relational
    (account-360 projects partner_of / supplies / competes_with edges into the same graph),
    and the embedded cypher() subset supports only an `id(<var>) = <key>` anchor — it CANNOT
    filter on an edge's `rel`. An untyped `MATCH (root)<-[*1..5]-(child)` walk here pulled a
    SUPPLIER into the corporate family the first time the suite ran on one bucket (found
    live: Globex, a supplier of Acme, inflated Acme's family forecast by $18k). So the family
    is a recursive CTE over `crm_edge` with an explicit rel filter — a top-level SELECT with
    an integer-literal root (edges are child->parent, so we climb e.dst -> e.src), feeding
    the ids into a SQL `IN (…)` (DuckDB has no `= ANY(array)`)."""
    kids = s.execute(text(
        "WITH RECURSIVE family(node) AS ("
        "  SELECT " + str(int(node_id)) + " "
        "  UNION "
        "  SELECT e.src FROM crm_edge e JOIN family f ON e.dst = f.node "
        "  WHERE e.rel = 'subsidiary_of') "
        "SELECT node FROM family"
    )).scalars().all()
    fam = list(dict.fromkeys([node_id, *[int(k) for k in kids]]))  # root first, de-duped
    ids = ",".join(str(int(i)) for i in fam)
    row = s.execute(text(
        "SELECT round(sum(o.amount), 2)                AS family_open, "
        "       round(sum(o.amount*ps.probability), 2) AS family_weighted, "
        "       count(*)                               AS deals "
        "FROM crm_node n "
        "JOIN opportunities o ON o.account_domain = n.biz_key AND o.status='open' "
        "JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage "
        f"WHERE n.kind='account' AND n.id IN ({ids})"
    )).mappings().one()
    return {"root": node_id, "family_node_ids": fam,
            "family_open": float(row["family_open"] or 0),
            "family_weighted": float(row["family_weighted"] or 0),
            "deals": int(row["deals"] or 0)}

Re-validated 2026-09-06 on the shared crm bucket — with account-360's partner/supplier edges present: GET /forecast/family/1 → family_node_ids [1, 2, 3], family_open 36000, family_weighted 15600, deals 2 — Globex and its supplier edge correctly excluded, Greyparrot's deals excluded, the $18k inflation gone.

Workflow op 4 — the risk gate (MODELS). POST /opportunities/{id}/risk runs the one route that leaves the bucket: it asks kimi-k2.6 whether an open deal's stage-default category still holds, then upserts the verdict to deal_risk (keyed on opportunity_id, so a re-assessment upserts — one row per deal). It carries the suite's forecast:override role gate (it downgrades a commit — see ## Auth), and its body teaches a second hard-won pattern: read / model / write, in three phases. A kimi-k2.6 reasoning turn can run for minutes (12s–170s observed live), and holding a checked-out pg connection across it is how this route died in validation — the idle connection dropped mid-call and the commit after the model reply failed. So the route snapshots what it read, releases the connection before the slow call, and writes in a fresh session after:

# routes.py — workflow op 5: the AI risk gate (Ignite Models kimi-k2.6)
@router.post("/opportunities/{opportunity_id}/risk")
def assess_risk(opportunity_id: str, q: RiskIn,
                user: dict = Depends(require_permission("forecast:override")),
                s: Session = Depends(db)):
    """READ / MODEL / WRITE, in three phases: a kimi-k2.6 reasoning turn can run for minutes,
    and holding a checked-out pg-wire connection across it is how this route died live (the
    idle DataK3 connection dropped mid-call; the commit after the model reply then failed).
    So we read and RELEASE the connection before the model call, and write in a FRESH session."""
    opp = s.get(Opportunity, opportunity_id)
    if not opp:
        raise HTTPException(404, "no such opportunity")
    if opp.status != "open":
        raise HTTPException(409, f"opportunity is status={opp.status!r}, not 'open' — the gate runs on open deals")
    # snapshot the fields first — rollback() expires ORM instances, and the model prompt must
    # not trigger a lazy refresh against a connection we just gave back.
    snap = SimpleNamespace(**{c.name: getattr(opp, c.name) for c in Opportunity.__table__.columns})
    s.rollback()
    s.close()                                     # give the connection back before the slow call
    v = _assess(_models_token(), snap, q.days_in_stage, q.last_activity_days)
    with SessionLocal() as s2:                    # a fresh connection for the write phase
        upsert(s2, DealRisk, [{
            "opportunity_id": opportunity_id, "forecast_category": v["forecast_category"],
            "risk": v["risk"], "days_in_stage": q.days_in_stage,
            "last_activity_days": q.last_activity_days, "reason": v["reason"],
            "assessed_at": _now(),
        }], key="opportunity_id")
        s2.commit()
    return {"ok": True, "opportunity_id": opportunity_id,
            "forecast_category": v["forecast_category"], "risk": v["risk"], "reason": v["reason"]}

Re-validated 2026-09-06 on the suite's book: assessing the stagnant opp-globex-poc returned {"forecast_category": "pipeline", "risk": "at_risk", "reason": "No activity in 41 days; stagnant prospect deal."} — a written, auditable verdict, upserted to one deal_risk row.

Adding a new business operation touches only routes.py (and maybe models.py) — the plumbing in db.py is fixed. The pattern is one Pydantic *In schema + one @router.<verb> function: write via upsert, typed graph via a recursive CTE, Models via a minted service-account token (see EXTENDING.md in the package). The deployed crm-forecaster (Step 6) is workflow op 2 on a schedule — pure SQL, which is why its service account needs nothing but k3.editor.

Auth — config at the edge, one gate in the app

On Ignite, end-user login is configuration, not code. The suite deploys with the crm-suite dodil-appid pool attached (user_pool: crm-suite in .dodil/deploy.yaml, issuer https://appid.dodil.io/ihdiash/crm-suite) and the per-cluster Ignite gateway runs the entire browser login at the edge — PKCE S256 against the pool issuer, an AEAD-sealed session cookie, JWT verification — then injects the verified identity: X-Dodil-User (sub, email, app_roles) and X-Dodil-User-Jwt (the raw verified token, carrying the catalog-expanded permissions claim). Inbound copies of those headers are stripped, so they can't be forged. The package's auth.py is therefore a header-trust reader, not a verifier — no JWKS client, no issuer/audience env — exposing current_user and require_permission for the one job the app keeps: role-based gating.

After the audit, this component kept exactly one gate: POST /opportunities/{id}/risk demands the forecast:override permission, because it can downgrade a commit — the verdict moves the number a sales leader reports upward. Everything else — the stage-model CRUD, the weighted read, even the recompute (deterministic and idempotent: running it twice writes the same rows the stage model dictates) — rides on the gateway's authentication alone. The suite's other surviving gates: orgs:qualify (lead-to-opportunity), leads:score (qualification-scoring), quotes:approve (quote-cpq) — all checked against the pool's sales / analyst / manager role catalog (manager carries forecast:override).

The two-plane rule is unchanged: the pool identifies the user; the app still reaches DataK3 via its own service account — an app-user is never a bucket principal. Locally, opt in to a stub identity with DEV_ALLOW_ANON=1; it carries no permissions unless you grant them (DEV_USER_PERMISSIONS=forecast:override), so the gate stays gated on a laptop. Pool creation, the redirect_uris allowlist, and the off-gateway verify-it-yourself path (iss and aud mandatory): App authentication; the catalog: App roles.

Get the code

The package is a real download — code/crm-pipeline-forecast/v1.tar. This post is a walkthrough of exactly those files; the tarball is source only (the image-mode crm-forecaster deploy is in Step 6):

models.py          # SQLAlchemy — pipeline_stages, forecast_snapshots, deal_risk + the crm/core masters (stubbed)
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app; CRUD + backfill +
                   #              weighted + recompute (idempotent) + family (typed CTE) +
                   #              risk (Models, forecast:override-gated)
db.py              # lazy engine (openapi() builds with no creds) + the ON CONFLICT upsert helper
auth.py            # gateway header-trust: current_user + require_permission — no verifier
sa_token.py        # deployed: mints + refreshes the service-account token for the pg-wire password
.env.example       # BUCKET + PG_HOST/PG_PORT + DODIL_TOKEN (or the SA pair) + DEV_ALLOW_ANON
requirements.txt   # sqlalchemy, psycopg[binary], pgvector, fastapi, uvicorn, pydantic, httpx

Run it — point .env at your bucket, create the tables from the models, serve:

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
 
cp .env.example .env      # then set DODIL_TOKEN (your `dodil auth login` token)
# BUCKET defaults to "crm"; create it once (Step 0), then create the tables from the models:
python -c "import db, models; models.Base.metadata.create_all(db.engine)"
 
uvicorn routes:app --reload
# POST /pipeline_stages, /opportunities, /accounts · POST /pipeline/backfill
# GET  /pipeline/weighted · POST /forecast/recompute · GET /forecast/family/{node_id}
# POST /opportunities/{id}/risk   (the gate — kimi-k2.6; needs the SA creds in .env)

models.Base.metadata.create_all is the ORM tab's every class at once — the same tables Steps 0–5 built by CLI, created from the natural-key models with no migration tool.

In the suite, this component doesn't run alone: the seven crm/* packages compose into one Ignite app — crm-suite-app mounts each component's APIRouter under a per-component prefix (this one at /pipeline-forecast) on one FastAPI process, over one engine to the one crm bucket, deployed through the git cycle with user_pool: crm-suite — see Ship a DODIL app. The shared bucket is exactly what surfaced the typed-traversal bug above — components validated in isolation compose on shared state, so re-validate on the composed bucket.

How the pillars map

One bucket, three pillars over one copy of the deals — no forecasting warehouse, no graph DB, no overnight sync.

JobThe usual stackOn DataK3
Stage model + probabilitiesA config screen in the CRMpipeline_stages rows you upsert
Weighted pipelineA BI report over a nightly extractSUM(amount×probability) over live rows
Commit / best-case / pipelineA $30k/yr forecasting seatforecast_snapshots, grouped in SQL
Family-weighted rollupCRM + a second graph database + a synccrm_graph — graph_khop JOINed to opportunities
"Is this commit real?"A rep's gutkimi-k2.6 risk gate → deal_risk, with a reason
Point your BI tool at itA warehouse connectordata connect — psql, DB = bucket

Customize — the decisions this skill asks you

Q1 · stages + probabilities — your sales process

"What are your pipeline stages, and the win probability at each?"

The two lists are parallel — stages[i] gets probabilities[i]. They seed pipeline_stages, and the probability is what drives every weighted number. Default is a 7-stage B2B process (prospect 0.1 → qualify 0.25 → demo 0.4 → proposal 0.6 → negotiate 0.8 → closed_won 1.0 / closed_lost 0.0). This is the core knob: change it and every forecast in the bucket re-weights on the next recompute.

Q2 · forecast_periods — monthly or quarterly?

"Do you forecast by month or by quarter?"

  • quarterly (default) → period key 2026-Q3; the snapshot grain the engine writes and re-upserts.
  • monthly → period key 2026-09. Finer grain, more snapshot rows, faster-moving number.

Sets the period-key format in crm-forecaster (the FORECAST_PERIODS env) and the ## Test assertion.

Q3 · risk_gate — deterministic only, or add the AI override? (default true)

"Should an AI flag stale/at-risk deals and override the forecast category?"

  • true (default) → Step 5 runs: kimi-k2.6 assesses open deals and writes deal_risk; the forecast can read the override instead of the stage default. Catches the untouched-commit case.
  • false → deterministic stage→category only; no Models call, no deal_risk writes. Pairs with a forecaster SA that is provably k3.editor-only (see Q4 / the thin path).

Q4 · family_rollup — roll up by corporate family? (default true)

"Do you want the weighted forecast rolled up across account hierarchies?"

  • true (default) → Step 4's graph_khop family rollup is built (needs crm/core's crm_graph).
  • false → per-owner/per-account forecast only; skips gracefully if crm/core ran hierarchy=false. This is the pure-SQL thin path: no graph, no gate — the forecaster is nothing but SQL rollups, and its service account needs only k3.editor.

Test

Every command below ran live against DataK3 (org IHDIASH) — the standalone demo values inline, and the reference package (models.py + the real routes.py route functions) re-validated end to end on 2026-09-06 on the persistent crm bucket as part of the composed suite: the weighted pipeline (open 63000 / weighted 20100, via-stage == via-column), the idempotent recompute (a second run re-writes the same snapshot rows), the typed subsidiary_of family rollup (node 1 → 36000 / 15600 / 2 deals, supplier Globex excluded), and the kimi-k2.6 gate (opp-globex-poc → at_risk, "No activity in 41 days; stagnant prospect deal."). The Step 6 Ignite deploy uses the image-mode pattern validated live 2026-09-02 on crm-lead-scorer (see the note below).

# 1 — the stage model is 7 rows and the probabilities are the param
dodil data sql -b "$BUCKET" "SELECT count(*) AS stage_count, round(sum(probability),2) AS prob_sum FROM pipeline_stages"
#  stage_count = 7, prob_sum = 3.15   (0.1+0.25+0.4+0.6+0.8+1.0+0.0)
 
# 2 — weighted pipeline is correct AND agrees across deal-column vs stage-JOIN
dodil data sql -b "$BUCKET" "
  SELECT round(sum(o.amount),2) AS open_amount,
         round(sum(o.amount*ps.probability),2) AS weighted_via_stage,
         round(sum(o.amount*o.probability),2)  AS weighted_via_col
  FROM opportunities o JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage
  WHERE o.status='open'"
#  open_amount = 74000, weighted_via_stage = 40400, weighted_via_col = 40400
 
# 3 — the period's forecast has commit / best_case / pipeline buckets
dodil data sql -b "$BUCKET" "SELECT forecast_category, weighted_amount FROM forecast_snapshots WHERE period='2026-Q3' ORDER BY forecast_category"
#  best_case 19200 | commit 19200 | pipeline 2000
 
# 4 — family-weighted rollup excludes Greyparrot (not in Acme's tree)
dodil data pg -b "$BUCKET" "
  SELECT round(sum(o.amount),2) AS family_open, round(sum(o.amount*ps.probability),2) AS family_weighted, count(*) AS deals
  FROM graph_khop('crm_graph', 1, 2, 'in') g
  JOIN crm_node n ON n.id=g.node AND n.kind='account'
  JOIN opportunities o ON o.account_domain=n.biz_key AND o.status='open'
  JOIN pipeline_stages ps ON ps.pipeline=o.pipeline AND ps.name=o.stage"
#  family_open = 36000, family_weighted = 26400, deals = 2   (Greyparrot's $30k excluded)
 
# 5 — the risk gate returns valid JSON and downgrades a stale commit
dodil ignite models chat kimi-k2.6 \
  --system 'Return ONLY JSON: {"forecast_category":"pipeline|best_case|commit|omitted|closed","risk":"on_track|slipping|at_risk","reason":"<=20 words"}. Downgrade a stale commit.' \
  --message 'stage: negotiate | default_forecast_category: commit | amount: 24000 | days_in_stage: 45 | last_activity_days: 45'
#  {"forecast_category":"best_case","risk":"at_risk","reason":"Untouched for 45 days in negotiation; stale commit downgraded."}
 
# 6 — re-running the snapshot roll is idempotent (same 3 rows, not 6)
dodil data sql -b "$BUCKET" "SELECT count(*) AS snapshot_rows FROM forecast_snapshots"
#  snapshot_rows = 3   (after a second upsert of the same period/owner/category keys)

Deploy note (one step) — image mode (Lane B), validated live 2026-09-02. The crm-forecaster deploy

  • POST /recompute in Step 6 uses image mode — the exact Lane B pattern (--code ./forecaster --dockerfile-path Dockerfile --port 8080 --health-path /healthz --allow-unauthenticated). Validated live 2026-09-02: this pattern deploys, serves /healthz + its route unauthenticated, and writes durably — confirmed end-to-end via the sibling crm-lead-scorer engine (written rows survived +154s, re-confirmed at +95s); crm-forecaster reuses that identical handler/deploy pattern. Note the deploy shape: image mode (the platform Kaniko-builds the Dockerfile on deploy — there is no --runtime python compile step), the handler is an HTTP server (/healthz + POST /recompute, not handler(payload, ctx)), it writes over the pg wire (psycopg, not a K3 HTTP API), and DODIL_SERVICE_ACCOUNT_ID is the cli-… serviceAccountId (the uuid fails client_credentials). The SQL it runs is the exact roll proven in assertions 1–4 above.

If dodil ignite app deploy returns IAM resource registration failed: broken pipe, deploy under a fresh app name — the half-created app can't be updated or deleted (UMA can't authorize an unregistered resource).

One-shot

Build sales-pipeline forecasting on DataK3 (one bucket = SQL + graph + Models). Confirm each step.
 
1. Bucket `crm`. If crm/core isn't present, stub the masters you consume:
   - opportunities (key opportunity_id): name, account_domain, primary_contact_email, pipeline, stage,
     status, amount(double), owner, source, close_date, created_at.
   - accounts (key org_domain): name, parent_domain, tier, country, industry, employee_band,
     annual_revenue(double). Seed the Acme family (acme.io + labs.acme.io + acme.eu) + Greyparrot, and 4
     open opportunities: opp-1 labs.acme.io negotiate $24000, opp-2 acme.eu proposal $12000,
     opp-3 greyparrot.ai demo $30000, opp-4 greyparrot.ai qualify $8000 (owner [email protected], pipeline sales).
   - Project accounts into crm_node (BIGINT KEY, ids 1-4) + crm_edge (subsidiary_of: 2->1, 3->1),
     then CREATE GRAPH crm_graph.
2. Create pipeline_stages (key stage_id) and seed 7 stages with probabilities [0.1,0.25,0.4,0.6,0.8,1.0,0.0]
   and forecast categories (prospect/qualify=pipeline, demo/proposal=best_case, negotiate=commit,
   closed_won=closed, closed_lost=omitted).
3. ALTER opportunities ADD forecast_category + probability; backfill each open deal from its stage.
4. Weighted pipeline = SUM(amount×probability) over open deals — prove deal-column and stage-JOIN agree ($40,400).
5. Create forecast_snapshots (key snapshot_id = period:pipeline:owner:category); recompute 2026-Q3/sales
   grouped by owner+forecast_category (commit/best_case/pipeline); upsert one row per bucket.
6. Family rollup: graph_khop('crm_graph',1,2,'in') JOIN opportunities JOIN pipeline_stages → Acme family
   weighted $26,400 (Greyparrot excluded).
7. Risk gate (kimi-k2.6): assess opp-1 (negotiate/commit, 45 days untouched) → JSON {forecast_category,
   risk, reason}; create deal_risk (key opportunity_id) and upsert the verdict.
8. Image-deploy crm-forecaster (own service account, k3.editor ONLY — pure SQL; Dockerfile HTTP server,
   port 8080, /healthz) that recomputes forecast_snapshots for a period; POST /recompute {period:2026-Q3}
   → {snapshots_written:3}.

Ship it — crm-forecaster on a schedule

crm-forecaster is an image-mode Ignite app — a Dockerfile the platform Kaniko-builds on deploy (Lane B build-on-deploy: no pre-pushed image, no pull secret), running an HTTP server that recomputes on each POST /recompute. Drive that from your own scheduler (cron → curl), since there's no server-side scheduler; keep it warm with --auto-min-instances 1 if a cold-start 502 on the first nightly hit would bite. The compact deploy is in Step 6; the full DODIL supply chain (git → CI → a scanned registry image → versioning and rollback) is walked end to end in Ship a DODIL App.

Connect your tools

Everything this build wrote lives in one DataK3 bucket, reachable by your own stack — not just the CLI. data connect crm prints the endpoints; point your tools straight at the same rows:

  • SQL over Postgres wire — psql, sqlx/diesel (Rust), psycopg/asyncpg (Python), node-postgres (TS).
  • Graph over Bolt — a Neo4j driver or cypher-shell against crm_graph for the family rollup.

Full, live-validated walkthrough: Connect your tools.

Conclusion

Sales forecasting — the weighted pipeline, the commit/best-case/pipeline buckets, the family rollup, and an AI check on the deals that look too good — is one DataK3 bucket: three merge-keyed tables, a graph JOIN over the same rows, and one Models call. The deterministic rollups are pure SQL (the engine's service account needs nothing but k3.editor); the model is the override, not the baseline. Change your process by editing pipeline_stages; re-run the roll and every number re-weights.

Next steps: