diff --git a/docs/architecture/adr-0015-beoordeling-escalation.md b/docs/architecture/adr-0015-beoordeling-escalation.md new file mode 100644 index 0000000..2b1a570 --- /dev/null +++ b/docs/architecture/adr-0015-beoordeling-escalation.md @@ -0,0 +1,77 @@ +# ADR-0015: Beoordeling escalation reassigns via an external-worker task + +- **Status:** Accepted +- **Date:** 2026-07-17 +- **Deciders:** Respellion engineering +- **Relates to:** S-14 (#15); proposal #98. Builds on ADR-0009 (external-task worker / Workflow + Client), ADR-0013 (behandel-portal wiring, the `Beoordelen` user task), ADR-0014 (the boundary-event + pattern on `Beoordelen`). + +## Context + +S-14 escalates a beoordeling that a behandelaar does not pick up in time: after 14 days the case must +move to the `teamlead` role (PRD §5, flow 5). The `Beoordelen` user task already exists, claimable by +the `behandelaar` candidate group; the teamlead role is seeded in the medewerker realm. + +Two forces shape this. + +1. **The task must stay open.** Escalation changes *who may claim* an unclaimed beoordeling, not the + work itself — so the timer must be **non-interrupting**: the `Beoordelen` task keeps running while + escalation happens alongside it. +2. **Reassigning an open task's candidate group needs code.** Flowable cannot rewrite the candidate + groups of an already-open user task from BPMN XML alone — that requires either a Java delegate/listener + embedded in the engine, or an out-of-process actor driving the REST API. The repository has held a + "stock Flowable image, no custom jars; the Workflow Client is the only code that talks to Flowable + (§8.2)" posture since ADR-0009. + +## Decision + +**A non-interrupting `P14D` boundary timer on `Beoordelen` fires an external-worker task +(`BeoordelingEscaleren`); the Workflow Client reassigns the still-open `Beoordelen` task from the +behandelaar group to teamlead.** + +- **Modelled in BPMN, driven by an external worker.** The timer routes a parallel token to an + `external-worker` service task on the `BeoordelingEscaleren` topic, ending at a dedicated "Beoordeling + geëscaleerd" end event. The model owns *when* escalation happens; the Workflow Client — the only code + that talks to Flowable (§8.2) — owns *how* the reassignment is applied, exactly as `OpenZaakAanmaken` + delegates the ZGW call (ADR-0009). No custom code runs inside Flowable. +- **Reassignment is a candidate-group swap.** The escalation worker finds the still-open `Beoordelen` + task in the escalating instance (task query by `processInstanceId` + `taskDefinitionKey`), adds + `teamlead` as a candidate group via the task identity links, then removes `behandelaar`. The task now + belongs to the teamlead; its history and variables are untouched. +- **Best-effort, mirroring beoordeling and withdrawal.** If the task is no longer open — the behandelaar + completed it in the window before the timer fired — the reassignment is a no-op. A failed reassignment + leaves the escalation job un-completed so Flowable redelivers it (§8.6), consistent with the + `OpenZaakAanmaken` worker. +- **Segregated interface.** The escalation methods live on `IBeoordelingEscalatieClient`, separate from + the `OpenZaakAanmaken` worker's `IExternalWorkerClient`, so the OpenZaak worker never sees escalation + (interface segregation). Both are implemented by the one `FlowableWorkflowClient`. + +## Consequences + +**Positive** + +- The escalation trigger is visible in `registratie.bpmn`; Flowable stays a stock image, and the + Workflow Client remains the sole Flowable client (§8.2 upheld, not bent). +- Reuses the external-worker mechanics (topic acquire/complete, hosted pump, per-tick scope, + redelivery-on-failure) wholesale — the new code is one client capability, one processor, one pump. +- Escalation latency is bounded by the worker's poll interval (seconds) — negligible against a 14-day + timer. + +**Negative / costs** + +- Escalation is two REST hops (add teamlead, remove behandelaar) rather than one atomic update; between + them the task is briefly claimable by both groups. Harmless at these volumes, and the pair is idempotent + on redelivery. +- The Flowable identity-link and management-job REST shapes are validated live (verify-domain fires the + timer early via the management API), not in the Workflow Client's unit tests, which stub the HTTP + exchange and assert only the request shape — consistent with ADR-0009 and ADR-0014. + +## Alternatives considered + +- **Flowable timer/task listener (Java delegate).** Reassign in-engine when the timer fires. Rejected: + it needs a custom jar in Flowable, breaking the stock-image, REST-only posture and adding a build/deploy + surface to the engine for no capability the external-worker route lacks. +- **Interrupting timer that re-creates the task for teamlead.** Cancel `Beoordelen` and start a fresh + teamlead task. Rejected: it loses the task's identity/history and complicates correlation, where a + candidate-group swap on the same task expresses "the same work, now the teamlead's" directly. diff --git a/docs/demo-script.md b/docs/demo-script.md index 5ece867..5c8b44c 100644 --- a/docs/demo-script.md +++ b/docs/demo-script.md @@ -298,3 +298,48 @@ interrupting boundary event ends it → the werkbak drops the case. > DigiD submit → trek aanvraag in → ingetrokken is the Playwright happy path > (`tests/e2e/withdrawal.spec.ts`); the owner-scoping + workflow cancellation are covered by the > `Een registratie intrekken` acceptance scenarios and the domain live check. + +## S-14 — Beoordeling escalation: 14 days unclaimed → teamlead (#15, ADR-0015) + +A beoordeling a behandelaar does not pick up within 14 days escalates to the teamlead. A +non-interrupting boundary timer on the `Beoordelen` task fires a `BeoordelingEscaleren` external task; +the domain's escalation worker reassigns the still-open task's candidate group from `behandelaar` to +`teamlead`, so it moves from the behandelaar werkbak into the teamlead's. The `Beoordelen` task keeps +its identity throughout — only who may claim it changes. + +The timer is 14 days, so the demo fires it early through Flowable's management API (exactly what the +verify-domain check automates): + +```bash +# 1. Submit at the self-service portal (http://localhost:8140/, jan-burger / test123). The case +# parks at Beoordelen, visible in the behandelaar werkbak (http://localhost:8142/, merel-behandelaar) +# but NOT claimed. +# +# 2. Find the parked instance and its Beoordelen task, then fire the boundary timer early: +FL=http://localhost:8090/flowable-rest/service +PID=$(curl -s -u rest-admin:test -X POST "$FL/query/tasks" -H 'Content-Type: application/json' \ + -d '{"processDefinitionKey":"registratie","taskDefinitionKey":"Beoordelen"}' \ + | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["processInstanceId"])') +TID=$(curl -s -u rest-admin:test -X POST "$FL/query/tasks" -H 'Content-Type: application/json' \ + -d '{"processDefinitionKey":"registratie","taskDefinitionKey":"Beoordelen"}' \ + | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])') +TJ=$(curl -s -u rest-admin:test "$FL/management/timer-jobs?processInstanceId=$PID" \ + | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])') +curl -s -u rest-admin:test -X POST "$FL/management/timer-jobs/$TJ" \ + -H 'Content-Type: application/json' -d '{"action":"move"}' +AJ=$(curl -s -u rest-admin:test "$FL/management/jobs?processInstanceId=$PID" \ + | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])') +curl -s -u rest-admin:test -X POST "$FL/management/jobs/$AJ" \ + -H 'Content-Type: application/json' -d '{"action":"execute"}' +# +# 3. Within a couple of poll cycles the task's candidate group flips to teamlead: +curl -s -u rest-admin:test "$FL/runtime/tasks/$TID/identitylinks" # → [{"group":"teamlead","type":"candidate"}] +``` + +**The path:** BPMN non-interrupting `P14D` boundary timer on `Beoordelen` → `BeoordelingEscaleren` +external task → domain escalation worker (`BeoordelingEscalatiePump`) → Workflow Client swaps the task's +candidate group behandelaar → teamlead (§8.2). + +> Both branches (escalate after 14 days; no-op when completed in time) are covered by the +> `Een beoordeling escaleren` acceptance scenarios and the Workflow Client unit tests; the timer firing +> and reassignment are asserted live by the verify-domain check. diff --git a/infra/run-domain-check.sh b/infra/run-domain-check.sh index 329209e..ef9f95d 100755 --- a/infra/run-domain-check.sh +++ b/infra/run-domain-check.sh @@ -157,4 +157,82 @@ for _ in $(seq 1 15); do done [ -n "$gone" ] || { echo "FAIL — Beoordelen task for $reg_id2 still active after withdrawal" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; } echo "OK — withdrawal cancelled the Beoordelen task; the registratie process ended (ingetrokken)" + +# ── S-14: escalation. A third registration parks at Beoordelen. We fire its 14-day boundary timer +# early via Flowable's management API (the timer job is moved to executable and run), which routes a +# parallel token to the BeoordelingEscaleren external task. The domain's escalation worker acquires +# it and reassigns the still-open Beoordelen task from the behandelaar group to teamlead (ADR-0015). ─ +echo ">> submitting a third registration to escalate" +loc3="$(docker run --rm --network "$net" curlimages/curl:latest \ + -fsS -D - -o /dev/null -X POST "http://$dom_ip:8080/registrations" \ + -H 'Content-Type: application/json' -d '{"bsn":"123456782"}' \ + | sed -n 's/\r$//; s/^[Ll]ocation: //p' | head -1)" +[ -n "$loc3" ] || { echo "FAIL — third POST /registrations returned no Location" >&2; exit 1; } +reg_id3="${loc3##*/}" +echo ">> third registration $reg_id3" + +# Extracts " " for a registration from a task-query response on stdin. +task_and_pid_for_reg() { REG_ID="$1" python3 -c "import os,sys,json +try: + d=json.load(sys.stdin) +except Exception: + d={} +rid=os.environ['REG_ID'] +t=next((t for t in (d.get('data') or []) + if any(v.get('name')=='registrationId' and v.get('value')==rid for v in (t.get('variables') or []))), None) +print(f\"{t['id']} {t['processInstanceId']}\" if t else '')"; } + +# The candidate groups on a task (space-separated, sorted) from a runtime identitylinks response. +candidate_groups() { python3 -c "import sys,json +try: + links=json.load(sys.stdin) +except Exception: + links=[] +print(' '.join(sorted(l.get('group') or '' for l in links if l.get('type')=='candidate' and l.get('group'))))"; } + +# The first job id in a management jobs/timer-jobs response on stdin. +first_job_id() { python3 -c "import sys,json +try: + d=json.load(sys.stdin) +except Exception: + d={} +print(((d.get('data') or [{}])[0]).get('id',''))"; } + +echo ">> polling Flowable for its Beoordelen task" +task_id3=""; pid3="" +for _ in $(seq 1 30); do + resp="$(flcurl -X POST "$fl_base/query/tasks" -H 'Content-Type: application/json' -d "$query" 2>/dev/null || true)" + read -r task_id3 pid3 <<<"$(printf '%s' "$resp" | task_and_pid_for_reg "$reg_id3")" + [ -n "$task_id3" ] && break + sleep 2 +done +[ -n "$task_id3" ] || { echo "FAIL — no Beoordelen task appeared for registration $reg_id3" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; } +echo ">> Beoordelen task $task_id3 (instance $pid3) is waiting for the behandelaar" + +echo ">> asserting the task starts out claimable by the behandelaar group" +before="$(flcurl "$fl_base/runtime/tasks/$task_id3/identitylinks" | candidate_groups)" +[ "$before" = "behandelaar" ] || { echo "FAIL — expected candidate group 'behandelaar', got '$before'" >&2; exit 1; } + +echo ">> firing the 14-day boundary timer early via the management API" +timer_id="$(flcurl "$fl_base/management/timer-jobs?processInstanceId=$pid3" | first_job_id)" +[ -n "$timer_id" ] || { echo "FAIL — no timer job found for instance $pid3" >&2; exit 1; } +# Move the timer job to an executable async job. Flowable's async executor (running in flowable-rest) +# then picks it up and fires the non-interrupting boundary event. It may run the job before we can +# look, so executing it explicitly is a best-effort nudge — tolerate the job already being gone. +flcurl -X POST "$fl_base/management/timer-jobs/$timer_id" -H 'Content-Type: application/json' -d '{"action":"move"}' >/dev/null +async_id="$(flcurl "$fl_base/management/jobs?processInstanceId=$pid3" 2>/dev/null | first_job_id || true)" +if [ -n "$async_id" ]; then + flcurl -X POST "$fl_base/management/jobs/$async_id" -H 'Content-Type: application/json' -d '{"action":"execute"}' >/dev/null 2>&1 || true +fi +echo ">> timer fired; the BeoordelingEscaleren token is parked for the domain worker" + +echo ">> polling until the escalation worker reassigns the beoordeling to the teamlead" +escalated="" +for _ in $(seq 1 30); do + groups="$(flcurl "$fl_base/runtime/tasks/$task_id3/identitylinks" 2>/dev/null | candidate_groups || true)" + [ "$groups" = "teamlead" ] && { escalated=1; break; } + sleep 2 +done +[ -n "$escalated" ] || { echo "FAIL — Beoordelen task not reassigned to teamlead (candidate groups: '$groups')" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; } +echo "OK — the 14-day timer escalated the still-open Beoordelen task to the teamlead" exit 0 diff --git a/services/domain/Big.Api/Program.cs b/services/domain/Big.Api/Program.cs index 46ed8fd..b6ee497 100644 --- a/services/domain/Big.Api/Program.cs +++ b/services/domain/Big.Api/Program.cs @@ -21,6 +21,7 @@ builder.Services.AddHttpClient(); builder.Services.AddTransient(sp => sp.GetRequiredService()); builder.Services.AddTransient(sp => sp.GetRequiredService()); builder.Services.AddTransient(sp => sp.GetRequiredService()); +builder.Services.AddTransient(sp => sp.GetRequiredService()); builder.Services.AddHttpClient(); builder.Services.AddScoped(); @@ -30,9 +31,13 @@ builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); +builder.Services.AddScoped(); // The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion. builder.Services.AddHostedService(); +// The escalation worker polls the BeoordelingEscaleren jobs the 14-day timer parks and reassigns +// each overdue beoordeling to the teamlead (S-14). +builder.Services.AddHostedService(); var app = builder.Build(); diff --git a/services/domain/Big.Application/Ports.cs b/services/domain/Big.Application/Ports.cs index 4a26dd5..8e74b72 100644 --- a/services/domain/Big.Application/Ports.cs +++ b/services/domain/Big.Application/Ports.cs @@ -86,3 +86,10 @@ public interface IRegistrationStore /// it) and the registration id it carries as a process variable. /// public sealed record OpenZaakJob(string JobId, RegistrationId RegistrationId); + +/// +/// An acquired BeoordelingEscaleren escalation job (S-14): the Flowable job id and the process +/// instance whose still-open Beoordelen task must be reassigned from behandelaar to teamlead +/// once the 14-day boundary timer fires (ADR-0015). +/// +public sealed record EscalatieJob(string JobId, string ProcessInstanceId); diff --git a/services/domain/Big.Infrastructure/BeoordelingEscalatieProcessor.cs b/services/domain/Big.Infrastructure/BeoordelingEscalatieProcessor.cs new file mode 100644 index 0000000..a2dac9f --- /dev/null +++ b/services/domain/Big.Infrastructure/BeoordelingEscalatieProcessor.cs @@ -0,0 +1,37 @@ +using Microsoft.Extensions.Logging; + +namespace Big.Infrastructure; + +/// +/// One poll tick of the beoordeling-escalation worker (S-14, ADR-0015): acquire the parked +/// BeoordelingEscaleren jobs — the tokens the 14-day boundary timer on Beoordelen spawns +/// — reassign each instance's still-open Beoordelen task to the teamlead, and complete the job. +/// A job that fails is logged and left un-completed so Flowable redelivers it (§8.6). Split out from +/// the hosted pump so the acquire→reassign→complete logic is unit-testable without a running host. +/// +public sealed class BeoordelingEscalatieProcessor( + IBeoordelingEscalatieClient client, + ILogger logger) +{ + /// Acquire and process up to escalations. Returns the number acquired. + public async Task PumpOnceAsync(int maxJobs, CancellationToken ct = default) + { + var jobs = await client.AcquireBeoordelingEscalatieJobsAsync(maxJobs, ct); + + foreach (var job in jobs) + { + try + { + await client.ReassignBeoordelingToTeamleadAsync(job.ProcessInstanceId, ct); + await client.CompleteBeoordelingEscalatieJobAsync(job.JobId, ct); + } + catch (Exception ex) + { + // Leave the job un-completed: its lock expires and Flowable redelivers it (§8.6). + logger.LogError(ex, "BeoordelingEscaleren job {JobId} failed; leaving it for redelivery.", job.JobId); + } + } + + return jobs.Count; + } +} diff --git a/services/domain/Big.Infrastructure/BeoordelingEscalatiePump.cs b/services/domain/Big.Infrastructure/BeoordelingEscalatiePump.cs new file mode 100644 index 0000000..9d444b2 --- /dev/null +++ b/services/domain/Big.Infrastructure/BeoordelingEscalatiePump.cs @@ -0,0 +1,49 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace Big.Infrastructure; + +/// +/// The hosted polling loop of the beoordeling-escalation worker (S-14, ADR-0015): on an interval it +/// resolves a scoped and asks it to drain the parked +/// BeoordelingEscaleren jobs. A deliberately thin shell — all acquire/reassign/complete logic +/// lives in the processor, which is unit-tested; this class only owns the timer, the per-tick scope, +/// and loop resilience. Structurally identical to . +/// +public sealed class BeoordelingEscalatiePump( + IServiceScopeFactory scopeFactory, + FlowableOptions options, + ILogger logger) : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + while (!stoppingToken.IsCancellationRequested) + { + try + { + using var scope = scopeFactory.CreateScope(); + var processor = scope.ServiceProvider.GetRequiredService(); + await processor.PumpOnceAsync(options.MaxJobsPerPoll, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + // A transient fault (e.g. Flowable briefly unreachable) must not kill the loop. + logger.LogError(ex, "BeoordelingEscaleren job poll failed; retrying after the poll interval."); + } + + try + { + await Task.Delay(options.PollInterval, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + } + } +} diff --git a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs b/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs index 6a723c8..d8933e2 100644 --- a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs +++ b/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs @@ -15,11 +15,14 @@ namespace Big.Infrastructure; /// The REST contract here is the one verified against a live flowable-rest engine (ADR-0009). /// public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions options) - : IWorkflowClient, IExternalWorkerClient, IUserTaskClient + : IWorkflowClient, IExternalWorkerClient, IUserTaskClient, IBeoordelingEscalatieClient { private const string Topic = "OpenZaakAanmaken"; + private const string EscalatieTopic = "BeoordelingEscaleren"; private const string ProcessDefinitionKey = "registratie"; private const string BeoordelenTaskKey = "Beoordelen"; + private const string BehandelaarGroup = "behandelaar"; + private const string TeamleadGroup = "teamlead"; private const string RegistrationIdVariable = "registrationId"; private const string ZaakUrlVariable = "zaakUrl"; private const string BesluitVariable = "besluit"; @@ -106,6 +109,47 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti response.EnsureSuccessStatusCode(); } + public async Task> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default) + { + var request = new AcquireJobsRequest(EscalatieTopic, options.LockDuration, maxJobs, options.WorkerId); + + var jobs = await PostAsync>( + "external-job-api/acquire/jobs", request, ct) ?? []; + + return [.. jobs.Select(job => new EscalatieJob(job.Id, job.ProcessInstanceId))]; + } + + public async Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default) + { + // The escalation token runs in parallel to the still-open Beoordelen task (non-interrupting + // boundary timer); find that task in this instance so we can move it to the teamlead. If the + // behandelaar completed it just before the timer fired there is nothing to reassign — a + // best-effort no-op (the timer/completion race, cf. §8.6). + var query = new TaskByInstanceQueryRequest(processInstanceId, BeoordelenTaskKey); + var page = await PostAsync( + "service/query/tasks", query, ct); + + var task = page?.Data?.FirstOrDefault(); + if (task is null) + return; + + // Add teamlead, then drop behandelaar: the task now belongs to the teamlead group. + using (var added = await SendAsync( + $"service/runtime/tasks/{task.Id}/identitylinks", + new IdentityLinkRequest(TeamleadGroup, "candidate"), ct)) + added.EnsureSuccessStatusCode(); + + await DeleteAsync( + $"service/runtime/tasks/{task.Id}/identitylinks/groups/{BehandelaarGroup}/candidate", ct); + } + + public async Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default) + { + using var response = await SendAsync( + $"external-job-api/acquire/jobs/{jobId}/complete", new CompleteJobRequest(options.WorkerId, []), ct); + response.EnsureSuccessStatusCode(); + } + private async Task GetAsync(string path, CancellationToken ct) { var message = new HttpRequestMessage(HttpMethod.Get, new Uri(options.BaseUrl, path)); @@ -115,6 +159,14 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti return await response.Content.ReadFromJsonAsync(ct); } + private async Task DeleteAsync(string path, CancellationToken ct) + { + var message = new HttpRequestMessage(HttpMethod.Delete, new Uri(options.BaseUrl, path)); + message.Headers.Authorization = new AuthenticationHeaderValue("Basic", BasicCredentials()); + using var response = await http.SendAsync(message, ct); + response.EnsureSuccessStatusCode(); + } + private async Task PostAsync(string path, TRequest body, CancellationToken ct) { using var response = await SendAsync(path, body, ct); @@ -154,6 +206,14 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti [property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey, [property: JsonPropertyName("includeProcessVariables")] bool IncludeProcessVariables); + private sealed record TaskByInstanceQueryRequest( + [property: JsonPropertyName("processInstanceId")] string ProcessInstanceId, + [property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey); + + private sealed record IdentityLinkRequest( + [property: JsonPropertyName("group")] string Group, + [property: JsonPropertyName("type")] string Type); + private sealed record ClaimTaskRequest( [property: JsonPropertyName("action")] string Action, [property: JsonPropertyName("assignee")] string Assignee); @@ -193,6 +253,10 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti private sealed record ExecutionDto([property: JsonPropertyName("id")] string Id); + private sealed record AcquiredEscalatieJob( + [property: JsonPropertyName("id")] string Id, + [property: JsonPropertyName("processInstanceId")] string ProcessInstanceId); + private sealed record AcquiredJob( [property: JsonPropertyName("id")] string Id, [property: JsonPropertyName("variables")] IReadOnlyList Variables) diff --git a/services/domain/Big.Infrastructure/IExternalWorkerClient.cs b/services/domain/Big.Infrastructure/IExternalWorkerClient.cs index c8fa2f6..ade8872 100644 --- a/services/domain/Big.Infrastructure/IExternalWorkerClient.cs +++ b/services/domain/Big.Infrastructure/IExternalWorkerClient.cs @@ -16,3 +16,23 @@ public interface IExternalWorkerClient /// Complete an acquired job, passing the opened zaak URL back into the process. Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default); } + +/// +/// The escalation side of the Workflow Client (S-14): the BeoordelingEscaleren external-worker +/// jobs parked by the 14-day boundary timer on Beoordelen, and the reassignment they drive. +/// Kept separate from (interface segregation) so the OpenZaak +/// worker never sees escalation. Implemented by — the only code +/// that talks to Flowable (§8.2, ADR-0015). +/// +public interface IBeoordelingEscalatieClient +{ + /// Acquire and lock up to BeoordelingEscaleren jobs. + Task> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default); + + /// Reassign the still-open Beoordelen task in the given process instance from the + /// behandelaar group to teamlead. Best-effort no-op if the task is no longer open. + Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default); + + /// Complete an acquired escalation job so its token reaches the escalation end event. + Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default); +} diff --git a/services/domain/Big.Tests/BeoordelingEscalatieProcessorTests.cs b/services/domain/Big.Tests/BeoordelingEscalatieProcessorTests.cs new file mode 100644 index 0000000..f39e10a --- /dev/null +++ b/services/domain/Big.Tests/BeoordelingEscalatieProcessorTests.cs @@ -0,0 +1,82 @@ +using Big.Application; +using Big.Infrastructure; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Big.Tests; + +// S-14 (#15): the escalation drain loop. Mirrors OpenZaakJobProcessor — acquire the parked +// BeoordelingEscaleren jobs, reassign each instance's Beoordelen task to teamlead, then complete the +// job. A job whose reassignment fails is logged and left un-completed for Flowable to redeliver (§8.6). +public class BeoordelingEscalatieProcessorTests +{ + /// A fake escalation client: scripts the jobs to acquire, records reassignments and + /// completions, and can be told to throw on reassigning a given instance. + private sealed class FakeEscalatieClient(params EscalatieJob[] jobs) : IBeoordelingEscalatieClient + { + public int AcquireCount { get; private set; } + public List Reassigned { get; } = []; + public List Completed { get; } = []; + public string? ThrowOnInstance { get; set; } + + public Task> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default) + { + AcquireCount++; + return Task.FromResult>(jobs.Take(maxJobs).ToList()); + } + + public Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default) + { + if (processInstanceId == ThrowOnInstance) + throw new InvalidOperationException("reassign failed"); + Reassigned.Add(processInstanceId); + return Task.CompletedTask; + } + + public Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default) + { + Completed.Add(jobId); + return Task.CompletedTask; + } + } + + [Fact] + public async Task Acquires_an_escalation_reassigns_to_teamlead_and_completes_the_job() + { + var client = new FakeEscalatieClient(new EscalatieJob("job-9", "pi-1")); + + var acquired = await new BeoordelingEscalatieProcessor( + client, NullLogger.Instance).PumpOnceAsync(5); + + Assert.Equal(1, acquired); + Assert.Equal("pi-1", Assert.Single(client.Reassigned)); + Assert.Equal("job-9", Assert.Single(client.Completed)); + } + + [Fact] + public async Task A_failing_reassign_is_left_uncompleted_for_flowable_to_redeliver() + { + var client = new FakeEscalatieClient(new EscalatieJob("job-9", "pi-1")) { ThrowOnInstance = "pi-1" }; + var logger = new CapturingLogger(); + + var acquired = await new BeoordelingEscalatieProcessor(client, logger).PumpOnceAsync(5); + + Assert.Equal(1, acquired); + Assert.Empty(client.Completed); + var error = Assert.Single(logger.Entries, e => e.Level == LogLevel.Error); + Assert.Contains("job-9", error.Message); + } + + [Fact] + public async Task Does_nothing_but_poll_when_there_are_no_escalations() + { + var client = new FakeEscalatieClient(); + + var acquired = await new BeoordelingEscalatieProcessor( + client, NullLogger.Instance).PumpOnceAsync(5); + + Assert.Equal(0, acquired); + Assert.Equal(1, client.AcquireCount); + Assert.Empty(client.Completed); + } +} diff --git a/services/domain/Big.Tests/FlowableWorkflowClientTests.cs b/services/domain/Big.Tests/FlowableWorkflowClientTests.cs index aff3bf3..d852e11 100644 --- a/services/domain/Big.Tests/FlowableWorkflowClientTests.cs +++ b/services/domain/Big.Tests/FlowableWorkflowClientTests.cs @@ -306,4 +306,108 @@ public class FlowableWorkflowClientTests await Assert.ThrowsAsync( () => client.CompleteBeoordelingAsync("task-1", BeoordelingsBesluit.Goedkeuren)); } + + // ── S-14 (#15): 14-day beoordeling escalation → reassign to teamlead (ADR-0015) ─────────────── + // The BPMN parks a parallel escalation token on a non-interrupting P14D boundary timer, surfaced as + // an external-worker job on the BeoordelingEscaleren topic. The worker reassigns the still-open + // Beoordelen task from the behandelaar group to teamlead, then completes the escalation job. + + [Fact] + public async Task Acquire_escalation_jobs_posts_the_escalation_topic_and_parses_the_process_instance() + { + var capture = new RequestCapture(); + var client = Client(capture.Responds(HttpStatusCode.OK, + """[{"id":"job-9","processInstanceId":"pi-1"}]""")); + + var jobs = await client.AcquireBeoordelingEscalatieJobsAsync(3); + + var job = Assert.Single(jobs); + Assert.Equal("job-9", job.JobId); + Assert.Equal("pi-1", job.ProcessInstanceId); + Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs", + capture.Seen!.RequestUri!.ToString()); + Assert.Contains("\"topic\":\"BeoordelingEscaleren\"", capture.Body); + Assert.Contains("\"numberOfTasks\":3", capture.Body); + Assert.Contains("\"workerId\":\"worker-x\"", capture.Body); + } + + [Theory] + [InlineData("[]")] + [InlineData("null")] + public async Task Acquire_escalation_jobs_returns_empty_when_none_are_parked(string body) + { + var capture = new RequestCapture(); + var client = Client(capture.Responds(HttpStatusCode.OK, body)); + + Assert.Empty(await client.AcquireBeoordelingEscalatieJobsAsync(1)); + Assert.NotNull(capture.Seen); + } + + [Fact] + public async Task Reassign_moves_the_open_beoordelen_task_from_behandelaar_to_teamlead() + { + var requests = new List<(HttpMethod Method, string Url, string? Body)>(); + var client = Client(new StubHandler(async req => + { + requests.Add((req.Method, req.RequestUri!.ToString(), + req.Content is null ? null : await req.Content.ReadAsStringAsync())); + // The task query returns the still-open Beoordelen task parked in this instance. + return req.RequestUri!.AbsoluteUri.EndsWith("service/query/tasks") + ? new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("""{"data":[{"id":"task-7"}],"total":1}""", + Encoding.UTF8, "application/json"), + } + : new HttpResponseMessage(HttpStatusCode.OK); + })); + + await client.ReassignBeoordelingToTeamleadAsync("pi-1"); + + // 1. Find the still-open Beoordelen task in this process instance. + var query = requests.Single(r => r.Url.EndsWith("service/query/tasks")); + Assert.Equal(HttpMethod.Post, query.Method); + Assert.Contains("\"processInstanceId\":\"pi-1\"", query.Body); + Assert.Contains("\"taskDefinitionKey\":\"Beoordelen\"", query.Body); + // 2. Add teamlead as a candidate group on that task. + var add = requests.Single(r => r.Method == HttpMethod.Post + && r.Url.EndsWith("service/runtime/tasks/task-7/identitylinks")); + Assert.Contains("\"group\":\"teamlead\"", add.Body); + Assert.Contains("\"type\":\"candidate\"", add.Body); + // 3. Remove behandelaar as a candidate group — the task now belongs to teamlead. + Assert.Contains(requests, r => r.Method == HttpMethod.Delete + && r.Url.EndsWith("service/runtime/tasks/task-7/identitylinks/groups/behandelaar/candidate")); + } + + [Fact] + public async Task Reassign_is_a_no_op_when_the_beoordelen_task_is_no_longer_open() + { + // The behandelaar completed it just before the timer fired: nothing to reassign, no throw. + var methods = new List(); + var client = Client(new StubHandler(req => + { + methods.Add(req.Method); + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("""{"data":[],"total":0}""", Encoding.UTF8, "application/json"), + }); + })); + + await client.ReassignBeoordelingToTeamleadAsync("pi-1"); + + Assert.DoesNotContain(HttpMethod.Delete, methods); + } + + [Fact] + public async Task Complete_escalation_job_posts_to_the_job_complete_endpoint() + { + var capture = new RequestCapture(); + var client = Client(capture.Responds(HttpStatusCode.NoContent)); + + await client.CompleteBeoordelingEscalatieJobAsync("job-9"); + + Assert.Equal(HttpMethod.Post, capture.Seen!.Method); + Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs/job-9/complete", + capture.Seen.RequestUri!.ToString()); + Assert.Contains("\"workerId\":\"worker-x\"", capture.Body); + } } diff --git a/services/domain/stryker-config.json b/services/domain/stryker-config.json index 935dfd7..0391844 100644 --- a/services/domain/stryker-config.json +++ b/services/domain/stryker-config.json @@ -4,7 +4,8 @@ "test-projects": ["Big.Tests/Big.Tests.csproj"], "reporters": ["progress", "html"], "mutate": [ - "!**/OpenZaakJobPump.cs" + "!**/OpenZaakJobPump.cs", + "!**/BeoordelingEscalatiePump.cs" ], "thresholds": { "high": 95, diff --git a/tests/acceptance/Acceptance.csproj b/tests/acceptance/Acceptance.csproj index 50d99ab..e062468 100644 --- a/tests/acceptance/Acceptance.csproj +++ b/tests/acceptance/Acceptance.csproj @@ -22,6 +22,9 @@ + + diff --git a/tests/acceptance/Features/EenBeoordelingEscaleren.feature b/tests/acceptance/Features/EenBeoordelingEscaleren.feature new file mode 100644 index 0000000..7072d43 --- /dev/null +++ b/tests/acceptance/Features/EenBeoordelingEscaleren.feature @@ -0,0 +1,22 @@ +# language: en +# Drives S-14 (#15). A beoordeling a behandelaar does not pick up within 14 days escalates to the +# teamlead: a non-interrupting boundary timer parks a BeoordelingEscaleren job (ADR-0015) which the +# escalation worker drains, reassigning the still-open Beoordelen task's candidate group to teamlead. +# A beoordeling completed before the timer fires does not escalate. This scenario exercises the +# escalation worker against an in-memory Flowable stand-in; the timer firing live is verify-domain. +Feature: Een beoordeling escaleren + Als teamleider wil ik dat een beoordeling die na 14 dagen niet is opgepakt naar mij escaleert + zodat aanvragen niet blijven liggen. + + Scenario: Na 14 dagen zonder oppakken escaleert de beoordeling naar de teamlead + Given a registration parked at the Beoordelen task for the behandelaar + When the 14-day escalation timer fires + And the escalation worker runs + Then the beoordeling is reassigned to the teamlead + + Scenario: Een tijdig afgeronde beoordeling escaleert niet + Given a registration parked at the Beoordelen task for the behandelaar + When the behandelaar completes the beoordeling before the timer fires + And the 14-day escalation timer fires + And the escalation worker runs + Then the beoordeling stays with the behandelaar diff --git a/tests/acceptance/Steps/EenBeoordelingEscalerenSteps.cs b/tests/acceptance/Steps/EenBeoordelingEscalerenSteps.cs new file mode 100644 index 0000000..1d49dee --- /dev/null +++ b/tests/acceptance/Steps/EenBeoordelingEscalerenSteps.cs @@ -0,0 +1,44 @@ +using Acceptance.Support; +using Big.Infrastructure; +using Microsoft.Extensions.Logging.Abstractions; +using Reqnroll; +using Xunit; + +namespace Acceptance.Steps; + +/// Bindings for EenBeoordelingEscaleren.feature (S-14). Drives the escalation worker +/// () against an in-memory Flowable stand-in; one instance +/// per scenario. Escalation has no domain-aggregate surface — it only reassigns who may claim the +/// still-open Beoordelen task — so the scenario asserts on the task's candidate group. +[Binding] +[Scope(Feature = "Een beoordeling escaleren")] +public sealed class EenBeoordelingEscalerenSteps +{ + private readonly InMemoryEscalatieClient _flowable = new(); + private string _processInstanceId = ""; + + [Given("a registration parked at the Beoordelen task for the behandelaar")] + public void GivenARegistrationParkedForTheBehandelaar() + => _processInstanceId = _flowable.ParkBeoordeling(); + + [When("the 14-day escalation timer fires")] + public void WhenTheEscalationTimerFires() + => _flowable.FireEscalationTimer(_processInstanceId); + + [When("the behandelaar completes the beoordeling before the timer fires")] + public void WhenTheBehandelaarCompletesBeforeTheTimer() + => _flowable.CompleteBeoordeling(_processInstanceId); + + [When("the escalation worker runs")] + public async Task WhenTheEscalationWorkerRuns() + => await new BeoordelingEscalatieProcessor( + _flowable, NullLogger.Instance).PumpOnceAsync(5); + + [Then("the beoordeling is reassigned to the teamlead")] + public void ThenTheBeoordelingIsReassignedToTheTeamlead() + => Assert.Equal("teamlead", _flowable.CandidateGroupFor(_processInstanceId)); + + [Then("the beoordeling stays with the behandelaar")] + public void ThenTheBeoordelingStaysWithTheBehandelaar() + => Assert.Equal("behandelaar", _flowable.CandidateGroupFor(_processInstanceId)); +} diff --git a/tests/acceptance/Support/InMemoryDomainPorts.cs b/tests/acceptance/Support/InMemoryDomainPorts.cs index 3e21752..4926b21 100644 --- a/tests/acceptance/Support/InMemoryDomainPorts.cs +++ b/tests/acceptance/Support/InMemoryDomainPorts.cs @@ -1,5 +1,6 @@ using Big.Application; using Big.Domain; +using Big.Infrastructure; namespace Acceptance.Support; @@ -73,6 +74,59 @@ public sealed class InMemoryUserTaskClient : IUserTaskClient } } +/// An in-memory Flowable stand-in for the beoordeling-escalation scenario (S-14): it models +/// one Beoordelen task per process instance — its candidate group and whether it is still open — and +/// the escalation jobs the non-interrupting 14-day boundary timer parks. It drives the escalation +/// worker's behaviour without a running Flowable; the timer firing live is the verify-domain check. +public sealed class InMemoryEscalatieClient : IBeoordelingEscalatieClient +{ + private sealed class ParkedTask + { + public string CandidateGroup { get; set; } = "behandelaar"; + public bool IsOpen { get; set; } = true; + } + + private readonly Dictionary _tasks = []; + private readonly List _parked = []; + private int _seq; + + /// A registration parks at Beoordelen, claimable by the behandelaar group. + public string ParkBeoordeling() + { + var pid = $"pi-{++_seq}"; + _tasks[pid] = new ParkedTask(); + return pid; + } + + /// The behandelaar completes the beoordeling before the timer fires: the task closes. + public void CompleteBeoordeling(string processInstanceId) => _tasks[processInstanceId].IsOpen = false; + + /// The 14-day boundary timer fires: a non-interrupting token parks an escalation job. + public void FireEscalationTimer(string processInstanceId) + => _parked.Add(new EscalatieJob($"job-{++_seq}", processInstanceId)); + + /// The candidate group that may currently pick the task up. + public string CandidateGroupFor(string processInstanceId) => _tasks[processInstanceId].CandidateGroup; + + public Task> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default) + => Task.FromResult>(_parked.Take(maxJobs).ToList()); + + public Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default) + { + // No-op if the behandelaar already completed it — the timer/completion race (§8.6). + var task = _tasks[processInstanceId]; + if (task.IsOpen) + task.CandidateGroup = "teamlead"; + return Task.CompletedTask; + } + + public Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default) + { + _parked.RemoveAll(j => j.JobId == jobId); + return Task.CompletedTask; + } +} + /// An in-memory registration store for the domain acceptance scenario. public sealed class InMemoryRegistrationStore : IRegistrationStore { diff --git a/workflows/registratie.bpmn b/workflows/registratie.bpmn index 99a7ec7..4817d99 100644 --- a/workflows/registratie.bpmn +++ b/workflows/registratie.bpmn @@ -15,7 +15,12 @@ S-11 adds withdrawal: while parked at Beoordelen the citizen can trek de aanvraag in — an interrupting message boundary event (RegistratieIngetrokken) cancels the task and ends the process via a dedicated "ingetrokken" end (ADR-0014). The Workflow Client delivers the message - to the task's execution; the BPMN owns the cancellation path. --> + to the task's execution; the BPMN owns the cancellation path. + S-14 adds escalation: a NON-interrupting boundary timer (P14D) on Beoordelen. If a behandelaar + has not picked the task up within 14 days it fires a parallel token to an external-worker task + (BeoordelingEscaleren); the Workflow Client reassigns the still-open Beoordelen task from the + behandelaar group to teamlead (ADR-0015). The Beoordelen task stays open throughout — the timer + only changes who may claim it. --> @@ -45,6 +50,25 @@ + + + + + P14D + + + + + + + + + + @@ -83,6 +107,24 @@ + + + + + + + + + + + + + + + + + +