# Agentic Scheduling (AI + Cron)
## What this builds
Scheduled pipelines that invoke `ai.text.generate` with no tools — one model call per
run — to analyze data, generate summaries, classify items, or make decisions. Results are
stored in Sekejap and optionally sent out over HTTP.
---
## Pipelines
1. `CRON every hour` → fetch recent events → AI summary → store
2. `CRON daily` → aggregate summaries → AI report → send to a webhook
3. `CRON every 15 min` → fetch unclassified queue items → AI classify → update rows
4. `GET /admin/reports` → list AI-generated summaries → render page
---
## Tables
```sql
CREATE TABLE events (_key TEXT PRIMARY KEY, ts INTEGER, kind TEXT, detail JSON)
CREATE TABLE ai_summaries (_key TEXT PRIMARY KEY, summary TEXT, patterns JSON, anomalies JSON, period TEXT, generated_at INTEGER)
CREATE TABLE incoming_queue (_key TEXT PRIMARY KEY, payload JSON, urgency TEXT, category TEXT, processed BOOLEAN)
```
`events` is assumed to be populated by whatever emits them (another pipeline,
an import job); this doc only reads it.
---
## `ai.text.generate` — how it takes input
`--prompt` is the prompt, literal or `{{ expr }}`, resolved against the
payload; a long one goes after `--`. It is the only source of the prompt: the
node reads nothing else of the payload. Instructions that hold for every call
(format, tone, constraints) go in `--system-prompt`.
Its flags, from the definition:
The answer is `text: { value, verified, data? }`: `value` is the text, `data`
the parsed JSON when `--schema` was given and passed. With `--schema` there is
no `JSON.parse` step to write.
---
## DSL
### hourly-data-summary — fetch + AI summarize
```
| trigger.schedule --cron "0 * * * *"
| javascript.script.run -- "return { cutoff: Date.now() - 3600000 }"
| sekejap.query.run --param "1={{ input.script.cutoff }}" -- "SELECT * FROM events WHERE ts > $1"
| ai.text.generate --provider openai --credential my-llm --answer-only --system-prompt "You are an operations analyst." --schema '{"type":"object","required":["summary","patterns","anomalies"],"properties":{"summary":{"type":"string"},"patterns":{"type":"array"},"anomalies":{"type":"array"}}}' -- Summarize these events, count patterns, and flag anything unusual. Events: {{ input.query.rows }}
| javascript.script.run -- "const r = input.text.data; return { key: 'summary-' + Date.now(), summary: r.summary, patterns: r.patterns, anomalies: r.anomalies, period: 'hourly', generated_at: Date.now() }"
| sekejap.query.run --write --param "1={{ $nodes.n4.script.key }}" --param "2={{ $nodes.n4.script.summary }}" --param "3={{ $nodes.n4.script.patterns }}" --param "4={{ $nodes.n4.script.anomalies }}" --param "5={{ $nodes.n4.script.period }}" --param "6={{ $nodes.n4.script.generated_at }}" -- "INSERT INTO ai_summaries (_key, summary, patterns, anomalies, period, generated_at) VALUES ($1, $2, $3, $4, $5, $6)"
```
### daily-metrics-report — aggregate + report + send
```
| trigger.schedule --cron "0 8 * * *"
| javascript.script.run -- "return { cutoff: Date.now() - 86400000 }"
| sekejap.query.run --param "1={{ input.script.cutoff }}" -- "SELECT * FROM ai_summaries WHERE generated_at > $1"
| ai.text.generate --provider openai --credential my-llm --answer-only --system-prompt "Reply with the markdown report only, no commentary." -- Write a daily operations report in markdown from these hourly summaries — executive summary, key metrics, trends, recommendations. Summaries: {{ input.query.rows }}
| kv.entry.get --key report_webhook_url --durable
| http.response.fetch --url "{{ input.entry.value }}" --method POST --body "{{ { report: input.text.value } }}"
```
`kv.entry.get` adds `entry: { key, value, found }` to the payload alongside
`input.text` — nothing is lost. Set `report_webhook_url` once via
`kv.entry.put --key report_webhook_url --value "https://hooks.example.com/reports"`
(or a Settings page) before this pipeline runs; there is no `env` scope to read a
URL from.
### queue-classifier — AI classify and route
```
| trigger.schedule --cron "*/15 * * * *"
| sekejap.query.run -- "SELECT * FROM incoming_queue WHERE processed = false LIMIT 10"
| ai.text.generate --provider openai --credential my-llm --answer-only --schema '{"type":"array","items":{"type":"object","required":["key","urgency","category"],"properties":{"urgency":{"enum":["high","medium","low"]}}}}' -- Classify each item by urgency (high, medium, low) and category. Reply with a JSON array of { key, urgency, category }, where key is the _key of that item. Items: {{ input.query.rows }}
| logic.foreach --from "input.text.data"
| sekejap.query.run --write --param "1={{ $item.urgency }}" --param "2={{ $item.category }}" --param "3={{ $item.key }}" -- "UPDATE incoming_queue SET urgency = $1, category = $2, processed = true WHERE _key = $3"
```
`logic.foreach` fans out one run per array element; `$item` stays in scope
for every downstream node in that run even after `sekejap.query.run` replaces
`input`.
### admin-reports — view AI reports
```
| trigger.webhook --route /admin/reports --method GET
| sekejap.query.run -- "SELECT * FROM ai_summaries ORDER BY generated_at DESC LIMIT 30"
| javascript.script.run -- "return { reports: input.query.rows }"
| web.response.send --template pages/admin-reports.tsx
```
---
## Nodes Used
- `trigger.schedule` — cron-based scheduling (`0 * * * *` = hourly, `0 8 * * *` = daily 8am)
- `sekejap.query.run` — SQL against Sekejap; output is `query: { columns, rows, row_count, truncated }`, plus `rows_affected` with `--write`
- `javascript.script.run` — shape rows for insert; return value is added as `script`, the rest of the payload is kept
- `ai.text.generate` — analysis, classification, report generation; `--schema` returns checked JSON as `text.data`
- `logic.foreach` — one downstream run per classified item
- `kv.entry.get --key <k>` / `kv.entry.put --key <k> --value <JSON>` — hold the webhook URL (there is no `env` scope); `kv.entry.get` answers `entry: { key, value, found }`
- `http.response.fetch` — send the report to an external webhook
- `web.response.send` — admin reporting page
---
## Templates Needed
- `pages/admin-reports.tsx` — display AI-generated summaries with timestamps
> A script cannot set the response. It returns a value; the graph decides what
> happens next. Branch with `logic.if` and let `web.response.send` answer —
> `--status`, `--header` (`Location`, `Set-Cookie`), `--body`. See
> `help("pipeline/examples/webhook-restapi-postgres")` § Answering with a status.