diff --git a/CLAUDE.md b/CLAUDE.md index 7791a8a..1369b08 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -85,7 +85,7 @@ Extensions built on the public API only: providers `Anthropic` (Messages API), ` preset; `docs/mcp.md`); the `GoogleCalendar` feature extension (`calendar` tool, `/calendar_connect|status|disconnect`, one Google account per chat keyed by session scope, tokens in the chat workspace under `.volcanic/google-calendar/`; `docs/google-calendar.md`); and the `Scheduler` feature extension (a process-wide -`JobScheduler` on `TimeProvider`, `jobs.json` under `/scheduler/`, the per-chat `schedule` tool, watcher +`JobScheduler` on `TimeProvider`, `jobs.json` per chat in the workspace under `.volcanic/scheduler/`, the per-chat `schedule` tool, watcher scripts through `IProcessSpawnPolicy`; `docs/telegram-bot.md#scheduling`). The Claude Code provider is the most intricate piece (`docs/claude-code-provider.md`): diff --git a/docs/architecture.md b/docs/architecture.md index 8ca0bd4..d5a1384 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -22,7 +22,7 @@ directories. `GoogleCalendarConnections` (see [google-calendar.md](google-calendar.md)). `Scheduler` is a feature extension with a process-wide `JobScheduler` (scheduled prompts and watcher scripts -on a `TimeProvider` timer, jobs in `/scheduler/jobs.json`), a per-session `schedule` tool bound to a +on a `TimeProvider` timer, jobs in each target's workspace, `.volcanic/scheduler/jobs.json`), a per-session `schedule` tool bound to a host target, and an `IJobDelivery` the host implements (see [telegram-bot.md](telegram-bot.md#scheduling)). ## Vertical slices diff --git a/docs/docker.md b/docs/docker.md index d853002..7ea6ad0 100644 --- a/docs/docker.md +++ b/docs/docker.md @@ -134,8 +134,8 @@ The compose file binds them to directories under `/media/disk1/containers/volcan | Volume | Path | Contents | |---|---|---| -| `workspaces` | `/data/workspaces` | One directory per chat (`/data/workspaces/`): `AGENTS.md` or `CLAUDE.md`, `PERSONA.md`, `MEMORY.md`, `inbox/` (received files), `browser/` (screenshots, downloads), project-level `.volcanic/extensions/`, the Google Calendar token (`.volcanic/google-calendar/token.json`) | -| `agent` | `/data/agent` | Sessions (`sessions/telegram-/`), `settings.json`, `extensions/*.dll`, chat time zones (`telegram/chat-settings.json`), `scheduler/jobs.json` | +| `workspaces` | `/data/workspaces` | One directory per chat (`/data/workspaces/`): `AGENTS.md` or `CLAUDE.md`, `PERSONA.md`, `MEMORY.md`, `inbox/` (received files), `browser/` (screenshots, downloads), project-level `.volcanic/extensions/`, the Google Calendar token (`.volcanic/google-calendar/token.json`), scheduled jobs (`.volcanic/scheduler/jobs.json`) | +| `agent` | `/data/agent` | Sessions (`sessions/telegram-/`), `settings.json`, `extensions/*.dll`, chat time zones (`telegram/chat-settings.json`); a `scheduler/jobs.json` from before jobs were per chat is moved into the workspaces on start | | `claude` | `/data/claude` | The Claude login (credentials and onboarding state) | Bind mounts survive redeploys and deleting the resource. Docker creates missing host directories as root; the diff --git a/docs/superpowers/plans/2026-10-01-volcanic-secretary-data-migration.md b/docs/superpowers/plans/2026-10-01-volcanic-secretary-data-migration.md new file mode 100644 index 0000000..c85c927 --- /dev/null +++ b/docs/superpowers/plans/2026-10-01-volcanic-secretary-data-migration.md @@ -0,0 +1,166 @@ +# Plan: migrating VolcanicSecretary data into VolcanicAgenticCore + +Status: plan only, nothing implemented. The source inventory comes from the VolcanicSecretary expert, based on its +master at `a1b1d12`; its full write-up is `docs/2026-09-30-data-inventory-for-agenticcore.md` in that repo. The +target formats were checked against this repo's code. + +## 1. What exists on each side + +| | VolcanicSecretary (Coolify `j67efkcfjyek2c396utfaxg7`) | VolcanicAgenticCore (Coolify `r8b5tgty5v5svat9rnnojahb`) | +|---|---|---| +| Host root | `/media/disk1/containers/volcanic_secretary/` | `/media/disk1/containers/volcanic_secretary_core/` | +| Workspace | `workspace/` → `/workspace` (a single owner) | `workspaces//` → `/data/workspaces/` | +| Agent dir | `pi-agent/` → `/pi-agent` (Pi 0.87.1) | `agent/` → `/data/agent` | +| Claude login | `claude/` → `/root/.claude` | `claude/` → `/data/claude` (uid 1654) | +| Chat key | none. Owner only: `TELEGRAM_OWNER_ID`, which is also the chat id | Telegram chat id | +| DB / external state | none | none | + +`` below means the value of `TELEGRAM_OWNER_ID`. + +## 2. Mapping and the conversion each item needs + +| Source | Target | Conversion | +|---|---|---| +| `workspace/AGENTS.md` | `workspaces//AGENTS.md` | Copy. Optionally split the persona (the text below "Standing instructions…") into `PERSONA.md`. Port the code-side prompt in `src/agent/system-prompt.ts` into AGENTS.md if we want the same behaviour | +| `workspace/MEMORY.md` | `workspaces//MEMORY.md` | Copy | +| `workspace/inbox/*` | `workspaces//inbox/` | Copy | +| `workspace/scripts/*` | `workspaces//scripts/` | Copy. The relative paths in job commands still resolve | +| stray root files (reports, PDFs, …) | `workspaces//` | Copy | +| `workspace/fortress/` profiles | — | **Drop** (we have no Fortress; Playwright profiles aren't compatible). Logins have to be redone | +| `workspace/.secretary/google-token.json` | `workspaces//.volcanic/google-calendar/token.json` | **Convert** (§3.2) | +| `workspace/.secretary/jobs.json` | `workspaces//.volcanic/scheduler/jobs.json` | Copy; the scheduler converts it on start (§3.3) | +| `pi-agent/sessions/--workspace--/*.jsonl` | `agent/sessions/telegram-/.jsonl` | **Convert**; a verbatim copy won't load (§3.1) | +| env `SECRETARY_TIMEZONE` (probably Europe/Warsaw) | `agent/telegram/chat-settings.json` | Generate `{"version":1,"chats":{"":{"timeZone":"Europe/Warsaw"}}}` | +| `claude/.credentials.json` | `claude/.credentials.json` | Copy (optional: re-running `/claude-code-login` is just as easy) | +| `pi-agent/{npm,tmp,settings.json,crashes.json,keybindings.json}`, the rest of `claude/`, `*.tmp` | — | Drop | + +## 3. Conversions + +### 3.1 Sessions (Pi v3 → our v1). The expert's "copy verbatim" does not hold + +`FileSessionStore.Parse` deserialises every line strictly. A Pi file fails in five places: + +1. **Header.** Pi has `{"type":"session","version":3,"id":…,"parentSession":""}`. Ours needs + `{"version":1,"sessionId":…,"cwd":…,"timestamp":…,"parentSessionId"?}`. A missing `sessionId` is a `required` + failure. +2. **Message timestamps.** Pi writes `message.timestamp` as epoch milliseconds (a number). Ours is a `DateTimeOffset` + (ISO string). +3. **User content.** Pi allows `"content":"text"`. Ours needs an array of blocks. +4. **Entry types we don't have**: `usage`, `custom_message`, `context_edit`, `branch_summary`. These make the + polymorphic discriminator throw. `ListAsync` silently skips the whole file, and the chat starts empty. +5. **File name.** Ours must be `.jsonl` (`PathFor`), not `_.jsonl`. + +Converter (a one-off script, Python or C#), per file: + +- header → `sessionId` = the Pi uuid with dashes removed (32 hex), `cwd` = `/data/workspaces/`, `timestamp` kept, + `parentSessionId` = the uuid taken from the old `parentSession` path's file name. +- entry `id`/`parentId`/`timestamp` kept as they are (8 hex, ISO). Each file is already a straight chain, which our + leaf rule (last line) expects. +- `message`: number `timestamp` → ISO; string user content → `[{"type":"text","text":…}]`; drop `usage.totalTokens` + and `usage.cost` (or leave them if unknown members are ignored; check with a test load). `thinking` blocks: keep + `thinking` and `signature`. Signatures from another `claude` session may be rejected on replay; strip thinking + blocks if a replay test fails. +- `compaction`, `model_change`, `thinking_level_change`, `label`, `session_info`: keep; the shapes match. + `model_change`: rewrite provider `pi-claude-code-provider` → `claude-code` and model `opus` → the id we run, or just + drop the entries. +- `custom_message` → our `message` with a `custom` role, or drop it. `usage`, `context_edit`, `branch_summary`: drop + them, and **re-link the next entry's `parentId`** to the dropped entry's parent so the chain stays intact. If a + `firstKeptEntryId` points at a dropped entry, move it to the next kept entry. +- Set each file's mtime to its last entry's time. The newest file must be the live conversation, because a restart + resumes the most recently modified session. + +Only the newest file has to load well. The older files are pre-`/new` history: convert them all, but only the newest +one has to pass a test load. + +Tool-call history: the Secretary used Pi's `bash/read/edit/write/...` tools plus Fortress and calendar tools. Old +`toolCall`s that name tools we don't have are only replayed as history and never executed. Check that the Claude +Code provider tolerates unknown tool names when it replays the transcript. If it doesn't, start the migrated chat +with a fresh session that has a compaction summary built from the old one, instead of carrying the raw history. + +### 3.2 Google Calendar token + +The source is `{"refresh_token":"1//0e…"}`. Our `GoogleTokenStore` reads `refreshToken`: key case is ignored but +the underscore still matters, so a straight copy reads as "not connected". Write +`{"refreshToken":"…","scope":"https://www.googleapis.com/auth/calendar","account":"","timeZone":"Europe/Warsaw","connectedAt":""}` +with mode 0600 and directories 0700. + +Both apps use the device flow with a "TVs and Limited Input devices" client. The refresh token is bound to the +client, so **set our `GOOGLE_CLIENT_ID`/`GOOGLE_CLIENT_SECRET` to the Secretary's values**. With a different client, +the owner just re-runs `/calendar_connect`, which is also a fine fallback. + +### 3.3 Scheduled jobs + +**Implemented:** jobs are stored per chat in `/.volcanic/scheduler/jobs.json`. `JobStore` recognises the +Secretary format (a bare array) in a chat's file on load, converts it for that chat, rewrites the file in our format, +and keeps the original as `jobs.json.secretary-`. The migration step is just a copy. See `docs/telegram-bot.md#importing-volcanicsecretary-jobs`. + +The source is a JSON array with no target. Ours is `{"version":1,"jobs":[…]}` with `target` = `""`. + +| Source | Ours | +|---|---| +| `source.kind=text` (with `deliver` direct or agent) | `kind:"prompt"`, `text`. **Behaviour change:** "direct" used to send the text verbatim with no model call; now the agent writes it. Prefix the text with `Remind me: ` to match how legacy reminders load | +| `source.kind=script`, `deliver:"agent"` | `kind:"watcher"`, `command`, `timeoutSeconds` = `timeoutSec` | +| `source.kind=script`, `deliver:"direct"` | `kind:"watcher"`. **Behaviour change:** stdout now always goes through the agent | +| `schedule.at` / `schedule.cron` | Same keys. Our cron takes 5 fields or a macro (`@daily` passes as is). A 6-field cron with seconds `0` loses that field; other 6-field crons are dropped with a warning | +| `id` (5 characters) | Kept | +| `createdAt`, `lastRunAt`, `lastResult{ok,exitCode,stderrTail,at}` | Keep, adding `timedOut:false` | + +Watcher compatibility: + +- Ours runs `/bin/sh -c`; theirs ran `bash -c`. Set `Scheduler__Shell` to bash, or check the commands. +- Our env filter strips `*_API_KEY` and similar from scripts. Theirs passed `OPENAI_API_KEY` and others through. Grep + `scripts/` for env use and add `PROCESS_ENV_ALLOW` entries on purpose. +- Python dependencies: their image had a venv at `/opt/tilion`. Check the scripts' imports against our image. +- Past-due `at` jobs fire immediately on first start. Drop or reschedule them during migration. + +## 4. Configuration (Coolify env on our app) + +- `TELEGRAM_ALLOWED_USER_IDS` = `` (the owner first). +- `TELEGRAM_BOT_TOKEN`: **decision**. Reusing the Secretary's token keeps the chat and its history, but both bots + can't poll at once (409), so the cutover is stop-then-start. A new bot allows a parallel trial, but the owner gets + a new chat. +- `OPENAI_API_KEY`, `STT_MODEL=gpt-4o-transcribe`, `GOOGLE_CLIENT_ID`/`SECRET` (copied), `DEFAULT_TIMEZONE`/`TZ` + = Europe/Warsaw. +- Do not copy the stale `GOOGLE_REFRESH_TOKEN`, or any `ANTHROPIC_*` / `CLAUDE_CODE_OAUTH_TOKEN`. +- Model: theirs was `opus`; ours defaults to `claude-code/claude-opus-5-5`. + +## 5. Gaps (features that don't migrate as data) + +- **Fortress** stealth browser and its saved logins: no equivalent. Playwright MCP is the replacement; sites have to + be logged in again. +- Their commands `/status /new /stop /compact /connect_calendar /calendar_done` differ from ours (`/calendar_connect`, + `/jobs`, `/cancel`, `/timezone`, …). Tell the owner. +- Direct (no-LLM) job delivery, see §3.3. +- A per-turn date/time line: ours has a timestamp header plus an environment section, which is equivalent. + +## 6. Cutover runbook + +1. **Verify on the host** (the expert couldn't): the Coolify env of `j67efkcfjyek2c396utfaxg7`; `du -sh` and + `ls -la` of the three dirs; whether `pi-agent/auth.json`, `skills/` and `extensions/` exist; `jq .` of + `jobs.json` (cron syntax, commands); `docker volume ls | grep secretary` for early-September leftovers. +2. **Dry run.** Copy the source dirs to a scratch location and run the converter into a scratch + `volcanic_secretary_core` tree. Load it with a local `dotnet run` of the Telegram host (or a small test that calls + `FileSessionStore.ListAsync`/`LoadAsync`, the `JobStore` load and `GoogleTokenStore`). Fix whatever fails. +3. Set the env on our Coolify app (§4). +4. **Freeze**: stop the Secretary container through the Coolify API so no new messages, jobs or memory edits arrive. +5. Take a final copy of the source dirs (keep this as the backup and leave the source untouched). Run the converter + into `/media/disk1/containers/volcanic_secretary_core/`. `chown` is a no-op on NTFS. +6. Deploy our app (Coolify API, fire-and-forget). +7. **Smoke test** as the owner: `/claude-code-doctor`, ask "what do you remember about me?" (MEMORY plus session + resume), `/jobs`, `/calendar_status`, send a voice note, open an inbox file. +8. Rollback: stop ours and start the Secretary. Its data was never modified. +9. After about two weeks: delete the Secretary app and archive `/media/disk1/containers/volcanic_secretary/`. + +## 7. Effort + +- Converter script (sessions, jobs, token, settings) plus a test load: about half a day. +- Host checks, dry run and fixes: about half a day. +- Cutover: under an hour of downtime. +- Optional: a port of `system-prompt.ts` into AGENTS.md, and checking the watcher scripts. + +## 8. Decisions for the owner + +1. Reuse the Secretary's bot token (one chat) or a new bot (parallel trial)? +2. Carry the full session history, or only the newest session (or a summary of it)? +3. Accept that "direct" jobs and alerts now go through the agent? +4. Split the persona out of AGENTS.md into PERSONA.md? diff --git a/docs/superpowers/specs/2026-09-26-scheduler-design.md b/docs/superpowers/specs/2026-09-26-scheduler-design.md index 631a0f1..8a6f2b0 100644 --- a/docs/superpowers/specs/2026-09-26-scheduler-design.md +++ b/docs/superpowers/specs/2026-09-26-scheduler-design.md @@ -111,9 +111,13 @@ same chat. **after** its delivery completes, so a crash mid-fire re-fires it after a restart (at-least-once). The same holds for a graceful stop: a delivery cut off by shutdown isn't recorded, so the one-off stays in `jobs.json` and fires late on the next start. -- **Persistence is one central file:** `/scheduler/jobs.json` (by default `~/.volcanic/agent/scheduler/`). - The workspace is deliberately not used: the agent's `bash` can write anywhere there, and a central file is loaded - without scanning every workspace. The format is `{ "version": 1, "jobs": [...] }` in `VolcanicJson` camelCase. +- **Persistence is one file per target** (changed 2026-10-02; originally one central `/scheduler/jobs.json`): + `/.volcanic/scheduler/jobs.json`, found on start through `IJobDelivery.KnownTargets`. The folder owns the + jobs (a stored `target` is overridden). The agent's `bash` can write the file, but that's no more than the + `schedule` tool already allows, and the file is only read on start. The old central file + (`SchedulerOptions.LegacyStorePath`) is moved into the targets' files on start and renamed `.migrated-`. + A VolcanicSecretary file (a bare array) in a target's folder is converted for that target. The format is + `{ "version": 1, "jobs": [...] }` in `VolcanicJson` camelCase. - **Direct delivery is text, through `IChatChannel`.** The sibling spec `2026-09-26-telegram-attachments-design.md` adds `IChatChannel` (Abstractions, `Ui/`) with `SendTextAsync`/`SendFileAsync`. `IJobDelivery` stays, because it does what the channel doesn't (target routing, prompt runs, workspaces, deliverability), but the Telegram @@ -292,9 +296,9 @@ schedule(action: create|list|cancel, id?, name?, kind?: reminder|prompt|watcher, ### 6. Telegram host wiring - **Options**: a new `Scheduler` section bound to `SchedulerOptions`: `Enabled` (true), - `StorePath` (`/scheduler/jobs.json`), `DefaultScriptTimeout` (300 s), `MaxJobsPerTarget` (50) and + `LegacyStorePath` (`/scheduler/jobs.json`, the retired shared file), `DefaultScriptTimeout` (300 s), `MaxJobsPerTarget` (50) and `Shell`. There's no time zone and no scrub list here: zones are per chat (§7), the environment comes from - the process environment hook. An empty `StorePath` means `/scheduler/jobs.json`; `Validate` rejects a + the process environment hook. An empty `LegacyStorePath` means `/scheduler/jobs.json`; `Validate` rejects a `DefaultScriptTimeout` outside 1 s–1 h, `MaxJobsPerTarget` < 1 and empty `Shell` entries (exit 2). - **`Program.cs`** registers `SchedulerOptions`, `IScriptRunner` → `ScriptRunner`, `TimeProvider.System` (one `TryAddSingleton`, shared with the message-context and attachments specs), `JobScheduler`, diff --git a/docs/telegram-bot.md b/docs/telegram-bot.md index 5687fbc..1f8ac92 100644 --- a/docs/telegram-bot.md +++ b/docs/telegram-bot.md @@ -36,7 +36,7 @@ settings use the `GoogleCalendar` section, the scheduler the `Scheduler` section | `Telegram:ReconnectInitialDelay` / `ReconnectMaxDelay` | `1 s` / `60 s` | Backoff after lost connection | | `DEFAULT_TIMEZONE` / `Telegram:DefaultTimeZone` | — (the machine's zone) | IANA zone for chats that haven't set one with `/timezone`, e.g. `Europe/Berlin`. An unknown name stops startup. Also the default for the core (`AgenticCoreOptions.DefaultTimeZone`) | | `Scheduler:Enabled` | `true` | Scheduled prompts and watcher scripts (see [Scheduling](#scheduling)) | -| `Scheduler:StorePath` | `/scheduler/jobs.json` | One file for every chat's jobs, outside the workspaces | +| `Scheduler:LegacyStorePath` | `/scheduler/jobs.json` | The retired shared job file. Jobs live in each chat's `/.volcanic/scheduler/jobs.json`; on start the jobs found here move into their chats' files and this file is renamed `jobs.json.migrated-` | | `Scheduler:DefaultScriptTimeout` | `00:05:00` | Watcher timeout when a job sets none (1 s to 1 h) | | `Scheduler:MaxJobsPerTarget` | `50` | Jobs per chat | | `Scheduler:Shell` | `/bin/sh -c` (`cmd.exe /d /s /c` on Windows) | The watcher shell as a list, command appended; same as `bash` by default | @@ -178,10 +178,29 @@ After a restart, a one-off that came due while the bot was down is delivered as "This was due at …; the bot was offline." in the prompt. Recurring jobs don't catch up: missed occurrences are skipped. -Jobs are per chat: `/jobs` and the agent only see this chat's jobs. Watcher failure alerts aren't part of the +Jobs are per chat: each chat's jobs are stored in its workspace, in `.volcanic/scheduler/jobs.json`, and `/jobs` and +the agent only see this chat's jobs. The file is read on start and rewritten on every change, so edit it only while the +bot is stopped; a job in a chat's file always belongs to that chat. Watcher failure alerts aren't part of the conversation, so the agent can't see them later. With `Scheduler:Enabled=false`, `/jobs` and `/cancel` reply that scheduling is turned off, and `/timezone` still works. +#### Importing VolcanicSecretary jobs + +Copy VolcanicSecretary's `.secretary/jobs.json` to the chat's `.volcanic/scheduler/jobs.json` +(`/data/workspaces//.volcanic/scheduler/jobs.json` in the container) and start the bot. The scheduler +recognises the old format (a bare array without chat ids), converts it for the chat whose folder it is in, saves it in +the current format, and keeps the original next to it as `jobs.json.secretary-`. Each job keeps its id +(a new one if another chat already uses it), schedule and last result: + +- a `text` job becomes a prompt. With `deliver: direct` (sent as is, no model call) the text gets `Remind me: `, so the + agent passes it on, as it does for the retired reminder kind; +- a `script` job becomes a watcher with the same command and timeout. Its output now always goes to the agent; +- a 6-field cron with seconds `0` loses that field. Other jobs our scheduler can't express (seconds, invalid fields) + are dropped with a warning in the log. + +Copy the workspace's `scripts/` into the owner's workspace too: watcher commands such as `python scripts/check.py` run +from there. A one-off that is already due fires on the first start. + ### Google Calendar Each chat connects its own Google account with `/calendar_connect`, and the agent's `calendar` tool only uses that diff --git a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs index 37d179f..64d108a 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs @@ -12,6 +12,9 @@ public interface IJobDelivery /// Working directory for the target's scripts. A path computation: must not create or wake a session. string WorkspaceFor(string target); + /// Targets whose workspace may hold a job file (), read once on start. Must not create or wake a session. + IEnumerable KnownTargets(); + /// False when the target may no longer be served (e.g. removed from the allow-list): the job is skipped, not deleted. Must not create or wake a session. bool IsDeliverable(string target); diff --git a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs index daf1ce3..100052e 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs @@ -1,10 +1,12 @@ +using System.Globalization; using Microsoft.Extensions.Logging; using Volcanic.AgenticCore.Abstractions.Processes; namespace Volcanic.AgenticCore.Extensions.Scheduler; /// -/// The process-wide scheduler: loads jobs.json, arms one timer for the soonest job +/// The process-wide scheduler: loads every known target's jobs.json ( in its +/// workspace; saves rewrite only the changed target's file), arms one timer for the soonest job /// (capped at 1 h, re-armed on every tick, create, cancel, fire and zone change), and fires jobs through the host's /// . A fire is tracked in flight and never awaited by the tick; an occurrence that comes due /// while the job is still in flight is skipped. A watcher's script run touches only the runner, IsDeliverable, @@ -22,7 +24,7 @@ public sealed class JobScheduler : IAsyncDisposable private readonly ISecretRedactor _redactor; private readonly TimeProvider _time; private readonly ILogger _logger; - private readonly JobStore _store; + private readonly Dictionary _stores = new(StringComparer.Ordinal); private readonly SemaphoreSlim _saveGate = new(1, 1); private readonly Lock _lock = new(); private readonly Dictionary _jobs = new(StringComparer.Ordinal); @@ -38,18 +40,17 @@ public sealed class JobScheduler : IAsyncDisposable public JobScheduler(SchedulerOptions options, IScriptRunner runner, ISecretRedactor redactor, TimeProvider time, ILogger logger) { ArgumentNullException.ThrowIfNull(options); - if (string.IsNullOrWhiteSpace(options.StorePath)) throw new ArgumentException("SchedulerOptions.StorePath is required.", nameof(options)); _options = options; _runner = runner; _redactor = redactor; _time = time; _logger = logger; - _store = new JobStore(options.StorePath, time, logger); } public bool IsStarted { get { lock (_lock) return _delivery is not null && !_stopped; } } public DateTimeOffset Now => _time.GetUtcNow(); - public string StorePath => _store.Path; + /// How many jobs are loaded, across all targets. + public int Count { get { lock (_lock) return _jobs.Count; } } /// When the timer fires next (tests). internal DateTimeOffset? NextWakeAt { get { lock (_lock) return _wakeAt; } } @@ -60,11 +61,11 @@ public sealed class JobScheduler : IAsyncDisposable /// The latest queued tick, without the fires it started (tests; unlike it doesn't wait for held fires). internal Task PendingTick { get { lock (_lock) return _pendingTick; } } - /// Load the store, fire overdue one-offs (late), compute cron occurrences strictly after now (no catch-up), arm the timer. + /// Load every target's file (moving the retired shared file's jobs in first), fire overdue one-offs (late), compute cron occurrences strictly after now (no catch-up), arm the timer. public async Task StartAsync(IJobDelivery delivery, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(delivery); - var jobs = await _store.LoadAsync(cancellationToken).ConfigureAwait(false); + var jobs = await LoadAllAsync(delivery, cancellationToken).ConfigureAwait(false); var now = _time.GetUtcNow(); lock (_lock) { @@ -97,6 +98,90 @@ public sealed class JobScheduler : IAsyncDisposable Tick(); // overdue one-offs fire now } + // ---- Loading ---- + + /// + /// Every known target's jobs. The retired shared file's jobs are merged into their targets' files first (an id the + /// target already has is skipped, so an interrupted move is safe to repeat), then the shared file is renamed. An id + /// taken by another target gets a new one. Changed files are saved. + /// + private async Task> LoadAllAsync(IJobDelivery delivery, CancellationToken cancellationToken) + { + var byTarget = new Dictionary>(StringComparer.Ordinal); + var changed = new HashSet(StringComparer.Ordinal); + foreach (var target in delivery.KnownTargets().Distinct(StringComparer.Ordinal)) + await TargetJobsAsync(target).ConfigureAwait(false); + + var legacy = _options.LegacyStorePath; + if (!string.IsNullOrWhiteSpace(legacy) && File.Exists(legacy)) + { + var moved = 0; + foreach (var job in await new JobStore(legacy, _time, _logger).LoadAsync(cancellationToken).ConfigureAwait(false)) + { + var list = await TargetJobsAsync(job.Target).ConfigureAwait(false); + if (list is null || list.Any(j => j.Id == job.Id)) continue; + list.Add(job); + changed.Add(job.Target); + moved++; + } + await SaveAsync().ConfigureAwait(false); + if (File.Exists(legacy)) // not set aside as unreadable + { + var renamed = $"{legacy}.migrated-{_time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}"; + File.Move(legacy, renamed, overwrite: true); + _logger.LogInformation("Moved {Count} job(s) from the shared job file {Path} into the chats' files; it is now {Renamed}", moved, legacy, renamed); + } + } + + var ids = new HashSet(StringComparer.Ordinal); + foreach (var (target, list) in byTarget) + for (var i = 0; i < list.Count; i++) + { + if (ids.Add(list[i].Id)) continue; + var id = NewId(ids.Contains); + _logger.LogWarning("Job {Id} ({Name}) of target {Target}: the id is taken by another target; it is now {NewId}", list[i].Id, list[i].Name, target, id); + list[i] = list[i] with { Id = id }; + ids.Add(id); + changed.Add(target); + } + await SaveAsync().ConfigureAwait(false); + return [.. byTarget.Values.SelectMany(j => j)]; + + // Null when the target has no usable workspace (the host can't map it): its jobs are skipped. + async Task?> TargetJobsAsync(string target) + { + if (byTarget.TryGetValue(target, out var list)) return list; + JobStore store; + try + { + store = StoreFor(delivery, target); + } + catch (Exception ex) + { + _logger.LogError(ex, "Target {Target} has no workspace for its job file; its jobs are skipped", target); + return null; + } + return byTarget[target] = [.. await store.LoadAsync(cancellationToken).ConfigureAwait(false)]; + } + + async Task SaveAsync() + { + foreach (var target in changed) + await StoreFor(delivery, target).SaveAsync(byTarget[target], cancellationToken).ConfigureAwait(false); + changed.Clear(); + } + } + + private JobStore StoreFor(IJobDelivery delivery, string target) + { + lock (_lock) + { + if (!_stores.TryGetValue(target, out var store)) + _stores[target] = store = new JobStore(JobStore.PathIn(delivery.WorkspaceFor(target)), _time, _logger, target); + return store; + } + } + /// Disarm, kill running scripts, and wait at most for in-flight fires. public async Task StopAsync(CancellationToken cancellationToken) { @@ -158,7 +243,7 @@ public sealed class JobScheduler : IAsyncDisposable next = schedule.At ?? Schedules.NextCron(schedule.Cron!, now, zone); _jobs[created.Id] = new Entry(created) { Next = next }; } - await PersistAsync(cancellationToken).ConfigureAwait(false); + await PersistAsync(target, cancellationToken).ConfigureAwait(false); Arm(); return new JobView(created, next); } @@ -180,7 +265,7 @@ public sealed class JobScheduler : IAsyncDisposable _inFlight.TryGetValue(id, out running); } if (running is not null) await TryCancelAsync(running.Cancel).ConfigureAwait(false); - await PersistAsync(cancellationToken).ConfigureAwait(false); + await PersistAsync(target, cancellationToken).ConfigureAwait(false); Arm(); return removed; } @@ -356,7 +441,7 @@ public sealed class JobScheduler : IAsyncDisposable result = await RunWatcherAsync(job, delivery, zone, scriptToken).ConfigureAwait(false); break; } - if (result is not null) await RecordAsync(job.Id, result).ConfigureAwait(false); + if (result is not null) await RecordAsync(job, result).ConfigureAwait(false); } /// Null when the run or its delivery was cancelled (job cancelled or scheduler stopping): the outcome is dropped. @@ -423,17 +508,17 @@ public sealed class JobScheduler : IAsyncDisposable } /// Update and persist the run fields; a one-off is removed now that its fire has completed. - private async Task RecordAsync(string id, JobResult result) + private async Task RecordAsync(ScheduledJob job, JobResult result) { lock (_lock) { - if (!_jobs.TryGetValue(id, out var entry)) return; // cancelled meanwhile - if (entry.Job.Schedule.At is not null) _jobs.Remove(id); + if (!_jobs.TryGetValue(job.Id, out var entry)) return; // cancelled meanwhile + if (entry.Job.Schedule.At is not null) _jobs.Remove(job.Id); else entry.Job = entry.Job with { LastRunAt = result.At, LastResult = result }; } try { - await PersistAsync(CancellationToken.None).ConfigureAwait(false); + await PersistAsync(job.Target, CancellationToken.None).ConfigureAwait(false); } catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { @@ -442,15 +527,15 @@ public sealed class JobScheduler : IAsyncDisposable Arm(); } - /// Saves the current jobs. The snapshot is taken inside the gate, so a later save never writes older state. - private async Task PersistAsync(CancellationToken cancellationToken) + /// Saves the target's current jobs to its file. The snapshot is taken inside the gate, so a later save never writes older state. + private async Task PersistAsync(string target, CancellationToken cancellationToken) { await _saveGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { List snapshot; - lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)]; - await _store.SaveAsync(snapshot, cancellationToken).ConfigureAwait(false); + lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).Where(j => j.Target == target).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)]; + await StoreFor(_delivery!, target).SaveAsync(snapshot, cancellationToken).ConfigureAwait(false); } finally { @@ -475,11 +560,13 @@ public sealed class JobScheduler : IAsyncDisposable private JobResult? LastResult(string id) { lock (_lock) return _jobs.TryGetValue(id, out var e) ? e.Job.LastResult : null; } /// Under the lock. Not reused while a cancelled job with that id is still in flight. - private string NewId() + private string NewId() => NewId(id => _jobs.ContainsKey(id) || _inFlight.ContainsKey(id)); + + private static string NewId(Func taken) { string id; do id = new string(Random.Shared.GetItems(IdAlphabet.AsSpan(), 5)); - while (_jobs.ContainsKey(id) || _inFlight.ContainsKey(id)); + while (taken(id)); return id; } diff --git a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs index 57af1f5..04eb3ea 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs @@ -11,8 +11,14 @@ namespace Volcanic.AgenticCore.Extensions.Scheduler; /// (temp file + rename) through one semaphore. A corrupt file is set aside and loading starts empty, so a bad file never /// keeps the host down; invalid jobs (including entries that don't deserialise) are dropped with a warning. A job of the /// retired reminder kind is loaded as a prompt asking the agent to pass the reminder on. +/// +/// A chat's own file (, with ) owns its jobs: every entry gets that target, +/// whatever it stores. A VolcanicSecretary jobs.json there (a bare array without targets) is converted, saved in +/// this format and the original kept as jobs.json.secretary-<timestamp>. Without a target (the retired +/// shared file) entries keep their own targets and a VolcanicSecretary file is set aside. +/// /// -public sealed class JobStore(string path, TimeProvider time, ILogger logger) +public sealed class JobStore(string path, TimeProvider time, ILogger logger, string? target = null) { public const int Version = 1; private const string ReminderPrefix = "Remind me: "; @@ -20,14 +26,30 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger) public string Path => path; + /// A chat's job file: <workspace>/.volcanic/scheduler/jobs.json. + public static string PathIn(string workspace) => System.IO.Path.Combine(workspace, ".volcanic", "scheduler", "jobs.json"); + public async Task> LoadAsync(CancellationToken cancellationToken) { if (!File.Exists(path)) return []; var json = await File.ReadAllTextAsync(path, cancellationToken).ConfigureAwait(false); + JsonElement root; + try + { + using var document = JsonDocument.Parse(json); + root = document.RootElement.Clone(); + } + catch (JsonException) + { + root = default; + } + if (root.ValueKind == JsonValueKind.Array) + return await LoadSecretaryAsync(root, cancellationToken).ConfigureAwait(false); + StoredFile? file; try { - file = VolcanicJson.Deserialize(json); + file = root.ValueKind == JsonValueKind.Object ? VolcanicJson.Deserialize(root) : null; } catch (JsonException) { @@ -35,17 +57,36 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger) } if (file is not { Version: Version, Jobs: not null }) { - var aside = $"{path}.corrupt-{time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}"; - File.Move(path, aside, overwrite: true); - logger.LogError("The job store {Path} could not be read; it was moved to {Aside} and the scheduler starts empty", path, aside); + SetAside("could not be read"); return []; } + return ReadAll([.. file.Jobs.Select(Owned)]); + } + private async Task> LoadSecretaryAsync(JsonElement entries, CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(target)) + { + SetAside("is a VolcanicSecretary job file, which belongs in a chat's workspace (.volcanic/scheduler/jobs.json)"); + return []; + } + var source = entries.EnumerateArray().ToList(); + var jobs = ReadAll([.. source.Select(FromSecretary)]); + var backup = $"{path}.secretary-{Stamp()}"; + File.Copy(path, backup, overwrite: true); + await SaveAsync(jobs, cancellationToken).ConfigureAwait(false); + logger.LogInformation("Converted the VolcanicSecretary job file {Path} for target {Target}: kept {Kept} of {Total} jobs; the original is {Backup}", + path, target, jobs.Count, source.Count, backup); + return jobs; + } + + private List ReadAll(IReadOnlyList entries) + { var jobs = new List(); var ids = new HashSet(StringComparer.Ordinal); - for (var i = 0; i < file.Jobs.Count; i++) + for (var i = 0; i < entries.Count; i++) { - var (job, problem) = Read(file.Jobs[i]); + var (job, problem) = Read(entries[i]); problem ??= ids.Add(job!.Id) ? null : "duplicate id"; if (problem is not null) { @@ -57,6 +98,15 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger) return jobs; } + private void SetAside(string why) + { + var aside = $"{path}.corrupt-{Stamp()}"; + File.Move(path, aside, overwrite: true); + logger.LogError("The job store {Path} {Why}; it was moved to {Aside} and the scheduler starts empty", path, why, aside); + } + + private string Stamp() => time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture); + public async Task SaveAsync(IReadOnlyList jobs, CancellationToken cancellationToken) { await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false); @@ -96,6 +146,15 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger) return (job, JobValidation.CheckStored(job)); } + /// In a chat's file the entry belongs to that chat, whatever target it stores. + private JsonElement Owned(JsonElement entry) + { + if (string.IsNullOrWhiteSpace(target) || entry.ValueKind != JsonValueKind.Object) return entry; + var node = JsonNode.Parse(entry.GetRawText())!.AsObject(); + node["target"] = target; + return JsonSerializer.SerializeToElement(node); + } + /// A stored reminder (fixed text sent as is) becomes a prompt; other entries pass through unchanged. private static JsonElement FromReminder(JsonElement entry) { @@ -104,12 +163,75 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger) return entry; var node = JsonNode.Parse(entry.GetRawText())!.AsObject(); node["kind"] = "prompt"; - if (entry.TryGetProperty("text", out var text) && text.ValueKind == JsonValueKind.String && text.GetString() is { } body && !string.IsNullOrWhiteSpace(body) - && body.Length + ReminderPrefix.Length <= JobValidation.MaxTextLength) - node["text"] = ReminderPrefix + body; + if (entry.TryGetProperty("text", out var text) && Remind(text) is { } reminder) node["text"] = reminder; return JsonSerializer.SerializeToElement(node); } + /// The text of a reminder with , or null when it's empty or the prefix doesn't fit. + private static string? Remind(JsonElement text) => + text.ValueKind == JsonValueKind.String && text.GetString() is { } body && !string.IsNullOrWhiteSpace(body) + && body.Length + ReminderPrefix.Length <= JobValidation.MaxTextLength + ? ReminderPrefix + body + : null; + + /// + /// One VolcanicSecretary job as an entry of this format. source.kind text becomes a prompt (a + /// deliver: direct one, which sent its text as is, a reminder), script a watcher; a 6-field cron with + /// seconds 0 loses that field. Anything else is left for to reject. + /// + private JsonElement FromSecretary(JsonElement entry) + { + if (entry.ValueKind != JsonValueKind.Object) return entry; + var node = new JsonObject { ["target"] = target }; + foreach (var name in (string[])["id", "name", "createdAt", "lastRunAt"]) + if (entry.TryGetProperty(name, out var value)) node[name] = JsonNode.Parse(value.GetRawText()); + + if (entry.TryGetProperty("schedule", out var schedule)) + { + var copy = JsonNode.Parse(schedule.GetRawText()); + if (copy is JsonObject scheduleNode && scheduleNode["cron"] is JsonValue cron && cron.TryGetValue(out var expression)) + scheduleNode["cron"] = WithoutZeroSeconds(expression); + node["schedule"] = copy; + } + + if (entry.TryGetProperty("source", out var source) && source.ValueKind == JsonValueKind.Object + && source.TryGetProperty("kind", out var kind) && kind.ValueKind == JsonValueKind.String) + { + switch (kind.GetString()) + { + case "text": + node["kind"] = "prompt"; + if (source.TryGetProperty("text", out var text)) + { + var direct = !entry.TryGetProperty("deliver", out var deliver) || deliver.ValueKind != JsonValueKind.String || deliver.GetString() != "agent"; + node["text"] = (direct ? Remind(text) : null) ?? JsonNode.Parse(text.GetRawText()); + } + break; + case "script": + node["kind"] = "watcher"; + if (source.TryGetProperty("command", out var command)) node["command"] = JsonNode.Parse(command.GetRawText()); + if (source.TryGetProperty("timeoutSec", out var timeout)) node["timeoutSeconds"] = JsonNode.Parse(timeout.GetRawText()); + break; + default: + node["kind"] = kind.GetString(); + break; + } + } + + if (entry.TryGetProperty("lastResult", out var result) && JsonNode.Parse(result.GetRawText()) is JsonObject resultNode) + { + resultNode["timedOut"] ??= false; + node["lastResult"] = resultNode; + } + return JsonSerializer.SerializeToElement(node); + } + + private static string WithoutZeroSeconds(string cron) + { + var fields = cron.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries); + return fields.Length == 6 && fields[0] == "0" ? string.Join(' ', fields[1..]) : cron; + } + private sealed record JobFile(int Version, List Jobs); private sealed record StoredFile(int Version, List? Jobs); diff --git a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/SchedulerOptions.cs b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/SchedulerOptions.cs index eb7b71c..a07726f 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/SchedulerOptions.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/SchedulerOptions.cs @@ -7,8 +7,12 @@ public sealed class SchedulerOptions public bool Enabled { get; set; } = true; - /// jobs.json. Empty: the host fills under its agent directory. - public string? StorePath { get; set; } + /// + /// The retired shared jobs.json. Jobs are kept per chat (); on start the jobs + /// found here move into their chats' files and this file is renamed jobs.json.migrated-<timestamp>. + /// Empty: the host fills under its agent directory. + /// + public string? LegacyStorePath { get; set; } /// Watcher timeout when the job sets none. public TimeSpan DefaultScriptTimeout { get; set; } = TimeSpan.FromSeconds(300); @@ -19,7 +23,7 @@ public sealed class SchedulerOptions public List Shell { get; set; } = []; /// <agentDir>/scheduler/jobs.json. - public static string DefaultStorePath(string agentDir) => Path.Combine(agentDir, "scheduler", "jobs.json"); + public static string DefaultLegacyStorePath(string agentDir) => Path.Combine(agentDir, "scheduler", "jobs.json"); /// Configuration problems that must stop the host, empty when valid. public IReadOnlyList Validate() diff --git a/hosts/Volcanic.AgenticCore.Telegram/Program.cs b/hosts/Volcanic.AgenticCore.Telegram/Program.cs index f949471..b0dedd0 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Program.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Program.cs @@ -109,7 +109,7 @@ if (ClaudeInstallation.Locate(Environment.GetEnvironmentVariable) is { } claude) logger.LogInformation("Using Claude Code at {Path}, model {Model}", claude, options.Model); else logger.LogWarning("Claude Code (claude) was not found on PATH; requests will fail until it is installed and logged in. Send /claude-code-doctor to the bot for details."); -logger.LogInformation("Scheduler: {State}", scheduling.Enabled ? scheduling.StorePath : "off"); +logger.LogInformation("Scheduler: {State}", scheduling.Enabled ? "on, jobs in each chat's .volcanic/scheduler/jobs.json" : "off"); logger.LogInformation("Default time zone: {Zone}", TimeZoneIds.Name(options.ConfiguredTimeZone() ?? TimeZoneInfo.Local)); logger.LogInformation("Allowed users: {Users}. Workspaces under {Root}", string.Join(", ", options.AllowedUserIds), options.WorkspaceRoot); diff --git a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerHostedService.cs b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerHostedService.cs index fc52f2a..9fbd7d7 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerHostedService.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerHostedService.cs @@ -10,7 +10,7 @@ public sealed class SchedulerHostedService(JobScheduler scheduler, TelegramJobDe public async Task StartAsync(CancellationToken cancellationToken) { await scheduler.StartAsync(delivery, cancellationToken).ConfigureAwait(false); - logger.LogInformation("Scheduler started; jobs in {Path}", scheduler.StorePath); + logger.LogInformation("Scheduler started with {Count} job(s)", scheduler.Count); } public Task StopAsync(CancellationToken cancellationToken) => scheduler.StopAsync(cancellationToken); diff --git a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerSetup.cs b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerSetup.cs index 2796dcb..df6a135 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerSetup.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/SchedulerSetup.cs @@ -19,11 +19,11 @@ public static class SchedulerSetup new() { Command = "cancel", Description = "Cancel a scheduled job: /cancel " }, ]; - /// The Scheduler section; an empty StorePath becomes <agentDir>/scheduler/jobs.json. + /// The Scheduler section; an empty LegacyStorePath becomes <agentDir>/scheduler/jobs.json. public static SchedulerOptions Options(IConfiguration section, string agentDir) { var options = section.Get() ?? new SchedulerOptions(); - if (string.IsNullOrWhiteSpace(options.StorePath)) options.StorePath = SchedulerOptions.DefaultStorePath(agentDir); + if (string.IsNullOrWhiteSpace(options.LegacyStorePath)) options.LegacyStorePath = SchedulerOptions.DefaultLegacyStorePath(agentDir); return options; } diff --git a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs index 086f275..21662d1 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs @@ -55,6 +55,12 @@ public sealed class TelegramJobDelivery(ChatRegistry chats, ChatSettingsStore se public string WorkspaceFor(string target) => chats.WorkspaceFor(ChatIdOf(target)); + /// Every chat with a workspace directory (<WorkspaceRoot>/<chatId>). + public IEnumerable KnownTargets() => + Directory.Exists(options.WorkspaceRoot) + ? Directory.EnumerateDirectories(options.WorkspaceRoot).Select(Path.GetFileName).OfType().Where(name => TryChatIdOf(name, out _)).ToList() + : []; + /// Private chat id = user id: the chat is served while its user is allow-listed. public bool IsDeliverable(string target) => TryChatIdOf(target, out var chatId) && options.IsAllowed(chatId); diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobSchedulerTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobSchedulerTests.cs index 992254c..5059869 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobSchedulerTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobSchedulerTests.cs @@ -213,7 +213,8 @@ public sealed class JobSchedulerTests Assert.False(await h.Scheduler.CancelAsync("222", mine.Id, CancellationToken.None)); Assert.True(await h.Scheduler.CancelAsync("111", mine.Id, CancellationToken.None)); Assert.False(await h.Scheduler.CancelAsync("111", mine.Id, CancellationToken.None)); - Assert.Equal([theirs.Id], (await h.Store.LoadAsync(CancellationToken.None)).Select(j => j.Id)); + Assert.Empty(await h.Store.LoadAsync(CancellationToken.None)); + Assert.Equal([theirs.Id], (await h.StoreFor("222").LoadAsync(CancellationToken.None)).Select(j => j.Id)); // each target has its own file } [Fact] @@ -298,10 +299,8 @@ public sealed class JobSchedulerTests public async Task A_stored_job_whose_zone_lookup_throws_does_not_stop_the_start() { await using var h = new SchedulerHarness(); - await h.Store.SaveAsync([ - Stored("bad11", JobKind.Prompt, new JobSchedule(null, "*/10 * * * *"), target: "not-a-chat"), - Stored("good1", JobKind.Prompt, new JobSchedule(null, "*/10 * * * *")), - ], CancellationToken.None); + await h.StoreFor("not-a-chat").SaveAsync([Stored("bad11", JobKind.Prompt, new JobSchedule(null, "*/10 * * * *"), target: "not-a-chat")], CancellationToken.None); + await h.Store.SaveAsync([Stored("good1", JobKind.Prompt, new JobSchedule(null, "*/10 * * * *"))], CancellationToken.None); h.Delivery.BadTargets.Add("not-a-chat"); await h.StartAsync(); @@ -325,7 +324,7 @@ public sealed class JobSchedulerTests await h.Scheduler.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); Assert.Equal("k7m2q", Assert.Single(await h.Store.LoadAsync(CancellationToken.None)).Id); - var delivery = new FakeJobDelivery(); + var delivery = new FakeJobDelivery { WorkspaceRoot = h.Delivery.WorkspaceRoot }; await using (var restarted = new JobScheduler(h.Options, h.Runner, h.Redactor, h.Time, Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance)) { await restarted.StartAsync(delivery, CancellationToken.None); @@ -427,4 +426,40 @@ public sealed class JobSchedulerTests Assert.Equal(TimeSpan.FromSeconds(5), requests[1].Timeout); Assert.Equal("Europe/Berlin", requests[1].TimeZoneId); } + + [Fact] + public async Task The_shared_file_is_moved_into_each_targets_file_on_start() + { + await using var h = new SchedulerHarness(); + await h.StoreFor("222").SaveAsync([Stored("both1", JobKind.Prompt, new JobSchedule(null, "@daily"), target: "222")], CancellationToken.None); + await h.LegacyStore.SaveAsync([ + Stored("mine1", JobKind.Prompt, new JobSchedule(null, "@daily")), + Stored("their", JobKind.Prompt, new JobSchedule(null, "@daily"), target: "222"), + Stored("both1", JobKind.Prompt, new JobSchedule(null, "@daily"), target: "222"), // already moved by an interrupted start + ], CancellationToken.None); + + await h.StartAsync(); + + Assert.Equal(["mine1"], (await h.Store.LoadAsync(CancellationToken.None)).Select(j => j.Id)); + Assert.Equal(["both1", "their"], (await h.StoreFor("222").LoadAsync(CancellationToken.None)).Select(j => j.Id).Order()); + Assert.Equal(3, h.Scheduler.Count); + Assert.False(File.Exists(h.Options.LegacyStorePath)); + Assert.Single(Directory.GetFiles(Path.GetDirectoryName(h.Options.LegacyStorePath)!, "jobs.json.migrated-*")); + } + + [Fact] + public async Task An_id_taken_by_another_target_is_renamed_on_start() + { + await using var h = new SchedulerHarness(); + await h.Store.SaveAsync([Stored("same1", JobKind.Prompt, new JobSchedule(null, "@daily"))], CancellationToken.None); + await h.StoreFor("222").SaveAsync([Stored("same1", JobKind.Prompt, new JobSchedule(null, "@daily"), target: "222")], CancellationToken.None); + + await h.StartAsync(); + + var ids = new[] { "111", "222" }.Select(t => Assert.Single(h.Scheduler.List(t)).Job.Id).ToList(); + Assert.Equal(2, ids.Distinct().Count()); + Assert.Contains("same1", ids); + var stored = (await h.Store.LoadAsync(CancellationToken.None)).Concat(await h.StoreFor("222").LoadAsync(CancellationToken.None)).Select(j => j.Id); + Assert.Equal(ids.Order(), stored.Order()); // the new id is saved + } } diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs index a25c45a..4dcb1fa 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs @@ -119,6 +119,104 @@ public sealed class JobStoreTests : IDisposable Assert.Equal("Remind me: Call the dentist", job.Text); } + private const string SecretaryFile = """ + [ + { + "id": "j7k2q", + "name": "Dentist reminder", + "schedule": { "at": "2026-10-01T09:00:00+02:00" }, + "source": { "kind": "text", "text": "Call the dentist" }, + "deliver": "direct", + "createdAt": "2026-09-28T17:41:02.118Z" + }, + { + "id": "m3n4p", + "name": "Morning brief", + "schedule": { "cron": "0 0 8 * * 1-5" }, + "source": { "kind": "text", "text": "Summarise my day" }, + "deliver": "agent", + "createdAt": "2026-09-28T17:41:02.118Z" + }, + { + "id": "w8r4t", + "name": "Price check", + "schedule": { "cron": "@daily" }, + "source": { "kind": "script", "command": "python scripts/check_price.py", "timeoutSec": 120 }, + "deliver": "direct", + "createdAt": "2026-09-28T17:41:02.118Z", + "lastRunAt": "2026-09-29T07:00:00.004Z", + "lastResult": { "ok": false, "exitCode": 1, "stderrTail": "Traceback ...", "at": "2026-09-29T07:00:00.004Z" } + }, + { + "id": "s9x2z", + "name": "Every 30 seconds", + "schedule": { "cron": "*/30 * * * * *" }, + "source": { "kind": "text", "text": "tick" }, + "deliver": "direct", + "createdAt": "2026-09-28T17:41:02.118Z" + } + ] + """; + + [Fact] + public async Task A_VolcanicSecretary_file_in_a_chats_folder_is_converted_in_place_for_that_chat() + { + Directory.CreateDirectory(Path.GetDirectoryName(StorePath)!); + await File.WriteAllTextAsync(StorePath, SecretaryFile); + + var jobs = await new JobStore(StorePath, _time, NullLogger.Instance, target: "111").LoadAsync(CancellationToken.None); + + Assert.Equal(["j7k2q", "m3n4p", "w8r4t"], jobs.Select(j => j.Id)); // 30-second cron can't be expressed: dropped + Assert.All(jobs, j => Assert.Equal("111", j.Target)); + + var direct = jobs[0]; + Assert.Equal(JobKind.Prompt, direct.Kind); + Assert.Equal("Remind me: Call the dentist", direct.Text); + Assert.Equal(new DateTimeOffset(2026, 10, 1, 9, 0, 0, TimeSpan.FromHours(2)), direct.Schedule.At); + Assert.Equal(new DateTimeOffset(2026, 9, 28, 17, 41, 2, 118, TimeSpan.Zero), direct.CreatedAt); + + var agent = jobs[1]; + Assert.Equal("Summarise my day", agent.Text); + Assert.Equal("0 8 * * 1-5", agent.Schedule.Cron); // seconds field 0 dropped + + var watcher = jobs[2]; + Assert.Equal(JobKind.Watcher, watcher.Kind); + Assert.Equal("python scripts/check_price.py", watcher.Command); + Assert.Equal(120, watcher.TimeoutSeconds); + Assert.Equal("@daily", watcher.Schedule.Cron); + Assert.Equal(new JobResult(false, 1, false, "Traceback ...", new DateTimeOffset(2026, 9, 29, 7, 0, 0, 4, TimeSpan.Zero)), watcher.LastResult); + + // The file is rewritten in the current format and the original is kept next to it. + Assert.Contains("\"version\": 1", await File.ReadAllTextAsync(StorePath)); + Assert.Equal(jobs, await new JobStore(StorePath, _time, NullLogger.Instance, target: "111").LoadAsync(CancellationToken.None)); + Assert.Equal(SecretaryFile, await File.ReadAllTextAsync(StorePath + ".secretary-20260926140300")); + } + + [Fact] + public async Task A_VolcanicSecretary_file_as_the_shared_file_is_set_aside() + { + Directory.CreateDirectory(Path.GetDirectoryName(StorePath)!); + await File.WriteAllTextAsync(StorePath, SecretaryFile); + + Assert.Empty(await Store().LoadAsync(CancellationToken.None)); + Assert.False(File.Exists(StorePath)); + Assert.Equal(SecretaryFile, await File.ReadAllTextAsync(StorePath + ".corrupt-20260926140300")); + } + + [Fact] + public async Task A_chats_file_owns_its_jobs_whatever_target_they_store() + { + await Store().SaveAsync([Prompt("k7m2q", target: "999")], CancellationToken.None); + + var job = Assert.Single(await new JobStore(StorePath, _time, NullLogger.Instance, target: "111").LoadAsync(CancellationToken.None)); + + Assert.Equal("111", job.Target); + } + + [Fact] + public void A_chats_file_lives_in_its_workspace() => + Assert.Equal(Path.Combine("/ws/111", ".volcanic", "scheduler", "jobs.json"), JobStore.PathIn("/ws/111")); + [Fact] public async Task Concurrent_saves_are_serialised() { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/Support/SchedulerFakes.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/Support/SchedulerFakes.cs index da3615c..b7a90c4 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/Support/SchedulerFakes.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/Support/SchedulerFakes.cs @@ -53,6 +53,9 @@ internal sealed class FakeJobDelivery : IJobDelivery return Path.Combine(WorkspaceRoot, target); } + public IEnumerable KnownTargets() => + Directory.Exists(WorkspaceRoot) ? Directory.EnumerateDirectories(WorkspaceRoot).Select(d => Path.GetFileName(d)) : []; + public bool IsDeliverable(string target) { lock (_lock) _calls.Add("IsDeliverable:" + target); @@ -127,7 +130,7 @@ internal sealed class SchedulerHarness : IAsyncDisposable public SchedulerHarness(DateTimeOffset? start = null, Action? configure = null) { Time = new FakeTimeProvider(start ?? Start); - Options = new SchedulerOptions { StorePath = Path.Combine(Root, "scheduler", "jobs.json") }; + Options = new SchedulerOptions { LegacyStorePath = Path.Combine(Root, "scheduler", "jobs.json") }; configure?.Invoke(Options); Delivery.WorkspaceRoot = Path.Combine(Root, "workspaces"); Scheduler = new JobScheduler(Options, Runner, Redactor, Time, NullLogger.Instance); @@ -140,7 +143,11 @@ internal sealed class SchedulerHarness : IAsyncDisposable public FakeScriptRunner Runner { get; } = new(); public FakeRedactor Redactor { get; } = new("hunter2-secret"); public JobScheduler Scheduler { get; } - public JobStore Store => new(Options.StorePath!, Time, NullLogger.Instance); + /// Target 111's file. + public JobStore Store => StoreFor("111"); + public JobStore StoreFor(string target) => new(JobStore.PathIn(Path.Combine(Delivery.WorkspaceRoot, target)), Time, NullLogger.Instance, target); + /// The retired shared file. + public JobStore LegacyStore => new(Options.LegacyStorePath!, Time, NullLogger.Instance); public Task StartAsync() => Scheduler.StartAsync(Delivery, CancellationToken.None); diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs index c5276a6..46a1642 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs @@ -187,7 +187,7 @@ public sealed class FormattingAndOptionsTests var options = Volcanic.AgenticCore.Telegram.Scheduling.SchedulerSetup.Options(config.GetSection("Scheduler"), "/agent"); Assert.True(options.Enabled); - Assert.Equal(Path.Combine("/agent", "scheduler", "jobs.json"), options.StorePath); + Assert.Equal(Path.Combine("/agent", "scheduler", "jobs.json"), options.LegacyStorePath); Assert.Equal(TimeSpan.FromSeconds(300), options.DefaultScriptTimeout); Assert.Equal(5, options.MaxJobsPerTarget); Assert.Empty(options.Validate()); diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs index 9dcf3c2..93ef9e7 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs @@ -90,7 +90,7 @@ public sealed class ChatHost : IAsyncDisposable Settings = new ChatSettingsStore(ChatSettingsStore.DefaultPath(agentDir), coreOptions.DefaultTimeZone ?? TimeZoneInfo.Local, NullLogger.Instance, clock); if (scheduler is not null) { - if (string.IsNullOrWhiteSpace(scheduler.StorePath)) scheduler.StorePath = SchedulerOptions.DefaultStorePath(agentDir); + if (string.IsNullOrWhiteSpace(scheduler.LegacyStorePath)) scheduler.LegacyStorePath = SchedulerOptions.DefaultLegacyStorePath(agentDir); Scheduler = new JobScheduler(scheduler, runner ?? new ScriptRunner(scheduler, _services.GetRequiredService(), NullLogger.Instance, clock), _services.GetRequiredService(), clock, NullLogger.Instance);