mirror of
https://github.com/arsvendg/Stirling-PDF.git
synced 2026-07-14 10:34:06 +02:00
Policies tidying (#6587)
# Description of Changes * Improve typing of API (breaking change but unreleased, frontend also updated in this PR) * Add ownership concept to policies * De-AI the comments * Update the `task dev:saas` rule to spawn the engine as well
This commit is contained in:
@@ -18,6 +18,15 @@ version: '3'
|
||||
tasks:
|
||||
dev:
|
||||
desc: "Start backend dev server"
|
||||
cmds:
|
||||
- task: dev:proprietary
|
||||
vars:
|
||||
PORT: '{{.PORT}}'
|
||||
AIENGINE_URL: '{{.AIENGINE_URL}}'
|
||||
AIENGINE_TIMEOUTSECONDS: '{{.AIENGINE_TIMEOUTSECONDS}}'
|
||||
|
||||
dev:proprietary:
|
||||
desc: "Start backend dev server in proprietary mode"
|
||||
ignore_error: true
|
||||
vars:
|
||||
PORT: '{{.PORT | default "8080"}}'
|
||||
@@ -50,9 +59,14 @@ tasks:
|
||||
PORT: '{{.PORT | default "8080"}}'
|
||||
# Override to "" to run the pure `saas` profile against your own SAAS_DB_*.
|
||||
PROFILES: '{{.PROFILES | default "dev"}}'
|
||||
AIENGINE_URL: '{{.AIENGINE_URL | default ""}}'
|
||||
AIENGINE_TIMEOUTSECONDS: '{{.AIENGINE_TIMEOUTSECONDS | default "120"}}'
|
||||
env:
|
||||
SERVER_PORT: '{{.PORT}}'
|
||||
STIRLING_FLAVOR: saas
|
||||
AIENGINE_URL: '{{.AIENGINE_URL}}'
|
||||
AIENGINE_ENABLED: '{{if .AIENGINE_URL}}true{{else}}false{{end}}'
|
||||
AIENGINE_TIMEOUTSECONDS: '{{.AIENGINE_TIMEOUTSECONDS}}'
|
||||
cmds:
|
||||
- cmd: cmd /c ".\gradlew.bat :stirling-pdf:bootRun {{if .PROFILES}}--args=\"--spring.profiles.include={{.PROFILES}}\"{{end}}"
|
||||
platforms: [windows]
|
||||
|
||||
+12
-16
@@ -60,24 +60,20 @@ tasks:
|
||||
|
||||
dev:saas:
|
||||
desc: "Start SaaS backend + frontend concurrently on free ports"
|
||||
vars:
|
||||
PORTS:
|
||||
sh: '{{if eq OS "windows"}}{{.FIND_FREE_PORT_PS}} 8080 5173{{else}}{{.FIND_FREE_PORT_SH}} 8080 5173{{end}}'
|
||||
BACKEND_PORT: '{{index (splitList "\n" .PORTS) 0}}'
|
||||
FRONTEND_PORT: '{{index (splitList "\n" .PORTS) 1}}'
|
||||
deps:
|
||||
- task: backend:dev:saas
|
||||
vars:
|
||||
PORT: '{{.BACKEND_PORT}}'
|
||||
- task: frontend:dev:saas
|
||||
vars:
|
||||
PORT: '{{.FRONTEND_PORT}}'
|
||||
BACKEND_URL: 'http://localhost:{{.BACKEND_PORT}}'
|
||||
OPEN: "true"
|
||||
cmds:
|
||||
- task: dev:_all
|
||||
vars: { FRONTEND: saas, BACKEND: saas }
|
||||
|
||||
dev:all:
|
||||
desc: "Start backend + frontend + engine concurrently on free ports"
|
||||
cmds:
|
||||
- task: dev:_all
|
||||
|
||||
dev:_all:
|
||||
internal: true
|
||||
vars:
|
||||
FRONTEND: '{{.FRONTEND | default "proprietary"}}'
|
||||
BACKEND: '{{.BACKEND | default "proprietary"}}'
|
||||
PORTS:
|
||||
sh: '{{if eq OS "windows"}}{{.FIND_FREE_PORT_PS}} 8080 5173 5001{{else}}{{.FIND_FREE_PORT_SH}} 8080 5173 5001{{end}}'
|
||||
BACKEND_PORT: '{{index (splitList "\n" .PORTS) 0}}'
|
||||
@@ -87,11 +83,11 @@ tasks:
|
||||
- task: engine:dev
|
||||
vars:
|
||||
PORT: '{{.ENGINE_PORT}}'
|
||||
- task: backend:dev
|
||||
- task: 'backend:dev:{{.BACKEND}}'
|
||||
vars:
|
||||
PORT: '{{.BACKEND_PORT}}'
|
||||
AIENGINE_URL: 'http://localhost:{{.ENGINE_PORT}}'
|
||||
- task: frontend:dev
|
||||
- task: 'frontend:dev:{{.FRONTEND}}'
|
||||
vars:
|
||||
PORT: '{{.FRONTEND_PORT}}'
|
||||
BACKEND_URL: 'http://localhost:{{.BACKEND_PORT}}'
|
||||
|
||||
+12
-26
@@ -13,26 +13,18 @@ import stirling.software.common.model.ApplicationProperties;
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* The single authority on which filesystem locations a policy may read from or write to. Folder
|
||||
* sources and sinks take a configured directory, so without this a user who can save a policy could
|
||||
* point one at Stirling's own config/secrets directory and exfiltrate (or overwrite) it. Every
|
||||
* folder source and sink runs its directory through {@link #requirePermitted(Path)} at save time
|
||||
* and again at run time.
|
||||
* Authority on which filesystem locations a policy may read/write. Checked at save time and again
|
||||
* at run time, fail-closed in order:
|
||||
*
|
||||
* <p>Enforced fail-closed, in order:
|
||||
* <ol>
|
||||
* <li>denied entirely under the {@code saas} profile;
|
||||
* <li>Stirling's own config dir always rejected, even if an allowed root were misconfigured to
|
||||
* contain it;
|
||||
* <li>must resolve within {@code policies.allowedFolderRoots}; none configured means all denied.
|
||||
* </ol>
|
||||
*
|
||||
* <ul>
|
||||
* <li><b>Disabled in SaaS</b> - folder access is never allowed when the {@code saas} profile is
|
||||
* active; a tenant must not reach the host filesystem at all.
|
||||
* <li><b>Protected paths</b> - Stirling's own config directory (settings, database, keys,
|
||||
* backups) is always rejected, even if an allowed root were misconfigured to contain it.
|
||||
* <li><b>Allowlist</b> - the directory must resolve within one of {@code
|
||||
* policies.allowedFolderRoots}; with none configured, all folder access is refused.
|
||||
* </ul>
|
||||
*
|
||||
* <p>Paths are compared after normalisation, so {@code ..} segments cannot walk out of an allowed
|
||||
* root. (Symlink escape is not defended here; an operator who configures an allowed root containing
|
||||
* a symlink to a sensitive location is trusted.)
|
||||
* <p>Compared after normalisation so {@code ..} cannot escape a root. Symlink escape is not
|
||||
* defended: an operator who roots an allowlist on a symlink to a sensitive location is trusted.
|
||||
*/
|
||||
@Component
|
||||
public class FolderAccessGuard {
|
||||
@@ -50,13 +42,7 @@ public class FolderAccessGuard {
|
||||
this.protectedRoots = List.of(normalize(Path.of(InstallationPathConfig.getConfigPath())));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check that {@code dir} is a permitted folder location, returning its normalised absolute
|
||||
* form.
|
||||
*
|
||||
* @throws IllegalArgumentException if folder access is disabled (SaaS or no roots configured),
|
||||
* the path is inside a protected directory, or it falls outside every allowed root
|
||||
*/
|
||||
/** Returns the normalised absolute path; throws if not permitted. */
|
||||
public Path requirePermitted(Path dir) {
|
||||
if (saasActive) {
|
||||
throw new IllegalArgumentException(
|
||||
@@ -81,7 +67,7 @@ public class FolderAccessGuard {
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/** Whether this policy reads from or writes to a folder, and so is subject to these rules. */
|
||||
/** Whether this policy touches a folder source/sink, and so is subject to these rules. */
|
||||
public boolean usesFolderAccess(Policy policy) {
|
||||
boolean readsFolder =
|
||||
policy.sources().stream().anyMatch(spec -> FOLDER_TYPE.equals(spec.type()));
|
||||
|
||||
+56
@@ -0,0 +1,56 @@
|
||||
package stirling.software.proprietary.policy.config;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
||||
import stirling.software.common.model.ApplicationProperties;
|
||||
import stirling.software.common.service.UserServiceInterface;
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* Decides who may act on a stored {@link Policy}: its owner and global admins only, with no
|
||||
* separate view/edit/run capability. Enforced only when login is enabled; single-user deployments
|
||||
* pass every check. The owner is assigned server-side, never from client input.
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class PolicyAccessGuard {
|
||||
|
||||
private final UserServiceInterface userService;
|
||||
private final ApplicationProperties applicationProperties;
|
||||
|
||||
/** Owner for a new policy: the current user, or {@code null} when login is disabled. */
|
||||
public String ownerForNewPolicy() {
|
||||
return enforced() ? userService.getCurrentUsername() : null;
|
||||
}
|
||||
|
||||
/** Whether the current user may view, edit, delete, or run the given stored policy. */
|
||||
public boolean canAccess(Policy policy) {
|
||||
if (!enforced() || userService.isCurrentUserAdmin()) {
|
||||
return true;
|
||||
}
|
||||
String current = userService.getCurrentUsername();
|
||||
return current != null && current.equals(policy.owner());
|
||||
}
|
||||
|
||||
/**
|
||||
* The subset of {@code policies} the current user may see (their own; everything for an admin).
|
||||
*/
|
||||
public List<Policy> visible(List<Policy> policies) {
|
||||
if (!enforced() || userService.isCurrentUserAdmin()) {
|
||||
return policies;
|
||||
}
|
||||
String current = userService.getCurrentUsername();
|
||||
if (current == null) {
|
||||
return List.of();
|
||||
}
|
||||
return policies.stream().filter(policy -> current.equals(policy.owner())).toList();
|
||||
}
|
||||
|
||||
private boolean enforced() {
|
||||
return applicationProperties.getSecurity().isEnableLogin();
|
||||
}
|
||||
}
|
||||
+29
@@ -0,0 +1,29 @@
|
||||
package stirling.software.proprietary.policy.controller;
|
||||
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
|
||||
import jakarta.validation.constraints.NotBlank;
|
||||
import jakarta.validation.constraints.NotNull;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* A supporting file paired with the asset key a pipeline step references from its {@code
|
||||
* fileParameters}. The same key may appear on more than one asset to supply multiple files.
|
||||
*/
|
||||
@Data
|
||||
@Schema(description = "A supporting file bound to the asset key a pipeline step references")
|
||||
public class NamedAsset {
|
||||
|
||||
@NotBlank
|
||||
@Schema(
|
||||
description = "Asset key referenced by a step's fileParameters",
|
||||
example = "company-logo")
|
||||
private String key;
|
||||
|
||||
@NotNull
|
||||
@Schema(description = "The supporting file", format = "binary")
|
||||
private MultipartFile file;
|
||||
}
|
||||
+107
-86
@@ -11,17 +11,16 @@ import org.springframework.core.io.Resource;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.util.MultiValueMap;
|
||||
import org.springframework.web.bind.annotation.DeleteMapping;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.ModelAttribute;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RequestPart;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
import org.springframework.web.multipart.MultipartHttpServletRequest;
|
||||
import org.springframework.web.server.ResponseStatusException;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
|
||||
@@ -30,6 +29,8 @@ import io.swagger.v3.oas.annotations.Hidden;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
|
||||
import jakarta.validation.Valid;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@@ -39,6 +40,7 @@ import stirling.software.common.service.UserServiceInterface;
|
||||
import stirling.software.common.util.TempFile;
|
||||
import stirling.software.common.util.TempFileManager;
|
||||
import stirling.software.proprietary.policy.config.FolderAccessGuard;
|
||||
import stirling.software.proprietary.policy.config.PolicyAccessGuard;
|
||||
import stirling.software.proprietary.policy.engine.PolicyRunHandle;
|
||||
import stirling.software.proprietary.policy.engine.PolicyRunRegistry;
|
||||
import stirling.software.proprietary.policy.engine.PolicyRunner;
|
||||
@@ -53,17 +55,9 @@ import stirling.software.proprietary.policy.progress.PolicyProgressListener;
|
||||
import stirling.software.proprietary.policy.store.PolicyStore;
|
||||
import stirling.software.proprietary.security.config.PremiumEndpoint;
|
||||
|
||||
import tools.jackson.core.JacksonException;
|
||||
import tools.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Manages policies and runs pipelines. The premium backend entry point: CRUD for stored {@code
|
||||
* Policy} objects, running a stored policy by id, and running an ad-hoc pipeline (for AI/Automate
|
||||
* one-offs).
|
||||
*
|
||||
* <p>Runs execute asynchronously and return a run id immediately. Poll {@code GET /run/{runId}} for
|
||||
* status, and download outputs via the existing {@code GET /api/v1/general/files/{fileId}} using
|
||||
* the file ids in the run view.
|
||||
* Policy CRUD plus pipeline runs (stored or ad-hoc). Runs are async: returns a run id, poll {@code
|
||||
* GET /run/{runId}} for status, download outputs via {@code GET /api/v1/general/files/{fileId}}.
|
||||
*/
|
||||
@Slf4j
|
||||
@RestController
|
||||
@@ -79,9 +73,9 @@ public class PolicyController {
|
||||
private final PolicyStore policyStore;
|
||||
private final PolicyValidator policyValidator;
|
||||
private final FolderAccessGuard folderAccessGuard;
|
||||
private final PolicyAccessGuard policyAccessGuard;
|
||||
private final UserServiceInterface userService;
|
||||
private final ApplicationProperties applicationProperties;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final TempFileManager tempFileManager;
|
||||
|
||||
@PostMapping(value = "/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
|
||||
@@ -89,15 +83,16 @@ public class PolicyController {
|
||||
summary = "Run a tool pipeline",
|
||||
description =
|
||||
"Accepts the documents to process (multipart field 'fileInput'), any supporting"
|
||||
+ " files (each under a multipart field named as its asset key, e.g."
|
||||
+ " 'company-logo'), and a JSON pipeline definition ('json'). Runs the"
|
||||
+ " steps in order asynchronously and returns a run id. Poll the run"
|
||||
+ " status endpoint and download outputs via /api/v1/general/files/{id}.")
|
||||
+ " files (under 'assets[i].key' / 'assets[i].file'), and the pipeline"
|
||||
+ " definition as an application/json part named 'json'. Runs the steps"
|
||||
+ " in order asynchronously and returns a run id. Poll the run status"
|
||||
+ " endpoint and download outputs via /api/v1/general/files/{id}.")
|
||||
public ResponseEntity<JobResponse<Void>> run(
|
||||
@RequestParam("json") String json, MultipartHttpServletRequest request)
|
||||
@RequestPart("json") PipelineDefinition definition,
|
||||
@Valid @ModelAttribute PolicyRunFiles files)
|
||||
throws IOException {
|
||||
PipelineDefinition definition = parseDefinition(json);
|
||||
PolicyInputs inputs = collectInputs(request);
|
||||
requireRunnable(definition);
|
||||
PolicyInputs inputs = toInputs(files);
|
||||
String runId =
|
||||
policyRunner.runAdHoc(definition, inputs, PolicyProgressListener.NOOP).runId();
|
||||
return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
|
||||
@@ -111,18 +106,19 @@ public class PolicyController {
|
||||
+ " starts and completes, then a terminal 'completed', 'failed',"
|
||||
+ " 'cancelled', or 'waiting' event carrying the final run view.")
|
||||
public SseEmitter runStream(
|
||||
@RequestParam("json") String json, MultipartHttpServletRequest request)
|
||||
@RequestPart("json") PipelineDefinition definition,
|
||||
@Valid @ModelAttribute PolicyRunFiles files)
|
||||
throws IOException {
|
||||
PipelineDefinition definition = parseDefinition(json);
|
||||
PolicyInputs inputs = collectInputs(request);
|
||||
requireRunnable(definition);
|
||||
PolicyInputs inputs = toInputs(files);
|
||||
|
||||
SseEmitter emitter =
|
||||
new SseEmitter(applicationProperties.getPolicies().getStreamTimeoutMs());
|
||||
emitter.onError(e -> log.warn("Policy run SSE emitter error", e));
|
||||
|
||||
PolicyRunHandle handle = policyRunner.runAdHoc(definition, inputs, streamListener(emitter));
|
||||
// Close the stream with a terminal event once the run finishes. whenComplete runs on the
|
||||
// engine's worker thread after the run is done, so this never races the step events.
|
||||
// whenComplete runs on the worker thread after the run finishes, so the terminal event
|
||||
// never races the step events.
|
||||
handle.completion()
|
||||
.whenComplete(
|
||||
(run, throwable) -> {
|
||||
@@ -159,22 +155,52 @@ public class PolicyController {
|
||||
description =
|
||||
"Stores a policy (trigger config + steps + output + metadata). A blank id is"
|
||||
+ " assigned; returns the stored policy with its id.")
|
||||
public ResponseEntity<Policy> savePolicy(@RequestBody String json) {
|
||||
Policy policy = parsePolicy(json);
|
||||
requireAuthorizedForFolderAccess(policy);
|
||||
public ResponseEntity<Policy> savePolicy(@RequestBody Policy policy) {
|
||||
Policy owned = resolveOwnership(policy);
|
||||
requireAuthorizedForFolderAccess(owned);
|
||||
try {
|
||||
policyValidator.validate(policy);
|
||||
policyValidator.validate(owned);
|
||||
} catch (IllegalArgumentException e) {
|
||||
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage());
|
||||
}
|
||||
return ResponseEntity.ok(policyStore.save(policy));
|
||||
return ResponseEntity.ok(policyStore.save(owned));
|
||||
}
|
||||
|
||||
/**
|
||||
* A policy that reads from or writes to a server folder grants whoever saves it access to that
|
||||
* path, so restrict it to administrators on multi-user deployments. Single-user deployments
|
||||
* (login disabled, e.g. desktop) trust the local operator. The {@link FolderAccessGuard} still
|
||||
* enforces SaaS-off and the path allowlist during validation regardless of who saves.
|
||||
* Assign the owner: create stamps the current user; update preserves the existing owner after
|
||||
* an access check. So the client can neither forge ownership on create nor reassign it on
|
||||
* update.
|
||||
*/
|
||||
private Policy resolveOwnership(Policy incoming) {
|
||||
String id = incoming.id();
|
||||
if (id != null && !id.isBlank()) {
|
||||
Policy existing = policyStore.get(id).orElse(null);
|
||||
if (existing != null) {
|
||||
if (!policyAccessGuard.canAccess(existing)) {
|
||||
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "No policy: " + id);
|
||||
}
|
||||
return withOwner(incoming, existing.owner());
|
||||
}
|
||||
}
|
||||
return withOwner(incoming, policyAccessGuard.ownerForNewPolicy());
|
||||
}
|
||||
|
||||
private static Policy withOwner(Policy policy, String owner) {
|
||||
return new Policy(
|
||||
policy.id(),
|
||||
policy.name(),
|
||||
owner,
|
||||
policy.enabled(),
|
||||
policy.trigger(),
|
||||
policy.sources(),
|
||||
policy.steps(),
|
||||
policy.output());
|
||||
}
|
||||
|
||||
/**
|
||||
* Folder sources/outputs grant whoever saves the policy access to that path, so gate them to
|
||||
* admins on multi-user deployments. Single-user (login disabled) trusts the local operator;
|
||||
* {@link FolderAccessGuard} still enforces SaaS-off and the path allowlist at validation time.
|
||||
*/
|
||||
private void requireAuthorizedForFolderAccess(Policy policy) {
|
||||
if (!folderAccessGuard.usesFolderAccess(policy)) {
|
||||
@@ -191,9 +217,11 @@ public class PolicyController {
|
||||
}
|
||||
|
||||
@GetMapping
|
||||
@Operation(summary = "List policies")
|
||||
@Operation(
|
||||
summary = "List policies",
|
||||
description = "Lists the caller's policies; admins see all.")
|
||||
public List<Policy> listPolicies() {
|
||||
return policyStore.all();
|
||||
return policyAccessGuard.visible(policyStore.all());
|
||||
}
|
||||
|
||||
@GetMapping("/{policyId}")
|
||||
@@ -201,6 +229,7 @@ public class PolicyController {
|
||||
public ResponseEntity<Policy> getPolicy(@PathVariable String policyId) {
|
||||
return policyStore
|
||||
.get(policyId)
|
||||
.filter(policyAccessGuard::canAccess)
|
||||
.map(ResponseEntity::ok)
|
||||
.orElseGet(() -> ResponseEntity.notFound().build());
|
||||
}
|
||||
@@ -208,9 +237,12 @@ public class PolicyController {
|
||||
@DeleteMapping("/{policyId}")
|
||||
@Operation(summary = "Delete a policy by id")
|
||||
public ResponseEntity<Void> deletePolicy(@PathVariable String policyId) {
|
||||
return policyStore.delete(policyId)
|
||||
? ResponseEntity.noContent().build()
|
||||
: ResponseEntity.notFound().build();
|
||||
boolean accessible =
|
||||
policyStore.get(policyId).filter(policyAccessGuard::canAccess).isPresent();
|
||||
if (accessible && policyStore.delete(policyId)) {
|
||||
return ResponseEntity.noContent().build();
|
||||
}
|
||||
return ResponseEntity.notFound().build();
|
||||
}
|
||||
|
||||
@PostMapping(value = "/{policyId}/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
|
||||
@@ -218,70 +250,52 @@ public class PolicyController {
|
||||
summary = "Run a stored policy",
|
||||
description =
|
||||
"Runs the stored policy's pipeline on the supplied files (primary documents"
|
||||
+ " under 'fileInput', supporting files under their asset-key fields)."
|
||||
+ " Runs regardless of the policy's enabled flag, which only gates"
|
||||
+ " automatic triggering. Returns a run id.")
|
||||
+ " under 'fileInput', supporting files under 'assets[i].key' /"
|
||||
+ " 'assets[i].file'). Runs regardless of the policy's enabled flag,"
|
||||
+ " which only gates automatic triggering. Returns a run id.")
|
||||
public ResponseEntity<JobResponse<Void>> runStoredPolicy(
|
||||
@PathVariable String policyId, MultipartHttpServletRequest request) throws IOException {
|
||||
@PathVariable String policyId, @Valid @ModelAttribute PolicyRunFiles files)
|
||||
throws IOException {
|
||||
Policy policy =
|
||||
policyStore
|
||||
.get(policyId)
|
||||
.filter(policyAccessGuard::canAccess)
|
||||
.orElseThrow(
|
||||
() ->
|
||||
new ResponseStatusException(
|
||||
HttpStatus.NOT_FOUND, "No policy: " + policyId));
|
||||
PolicyInputs inputs = collectInputs(request);
|
||||
PolicyInputs inputs = toInputs(files);
|
||||
String runId = policyRunner.runWith(policy, inputs, PolicyProgressListener.NOOP).runId();
|
||||
return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
|
||||
}
|
||||
|
||||
private Policy parsePolicy(String json) {
|
||||
try {
|
||||
return objectMapper.readValue(json, Policy.class);
|
||||
} catch (JacksonException e) {
|
||||
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Invalid policy JSON");
|
||||
}
|
||||
}
|
||||
|
||||
private PipelineDefinition parseDefinition(String json) {
|
||||
PipelineDefinition definition;
|
||||
try {
|
||||
definition = objectMapper.readValue(json, PipelineDefinition.class);
|
||||
} catch (JacksonException e) {
|
||||
throw new ResponseStatusException(
|
||||
HttpStatus.BAD_REQUEST, "Invalid pipeline definition JSON");
|
||||
}
|
||||
private static void requireRunnable(PipelineDefinition definition) {
|
||||
if (definition.steps().isEmpty()) {
|
||||
throw new ResponseStatusException(
|
||||
HttpStatus.BAD_REQUEST, "Pipeline definition has no steps");
|
||||
}
|
||||
return definition;
|
||||
}
|
||||
|
||||
/**
|
||||
* Split the multipart file parts into the primary document stream ("fileInput") and the named
|
||||
* supporting-file store: every other file field becomes an asset keyed by its field name, which
|
||||
* a step references from {@code fileParameters}.
|
||||
* Turn the typed run files into engine {@link PolicyInputs}: the primary documents plus the
|
||||
* named supporting-file store, where each asset's {@code key} is the name a step references
|
||||
* from its {@code fileParameters}. Assets sharing a key are grouped, so a key may carry several
|
||||
* files.
|
||||
*/
|
||||
private PolicyInputs collectInputs(MultipartHttpServletRequest request) throws IOException {
|
||||
MultiValueMap<String, MultipartFile> fileMap = request.getMultiFileMap();
|
||||
List<Resource> primary = toResources(fileMap.get("fileInput"));
|
||||
private PolicyInputs toInputs(PolicyRunFiles files) throws IOException {
|
||||
List<Resource> primary = toResources(files.getFileInput());
|
||||
Map<String, List<Resource>> supportingFiles = new LinkedHashMap<>();
|
||||
for (Map.Entry<String, List<MultipartFile>> entry : fileMap.entrySet()) {
|
||||
if ("fileInput".equals(entry.getKey())) {
|
||||
continue;
|
||||
}
|
||||
List<Resource> assets = toResources(entry.getValue());
|
||||
if (!assets.isEmpty()) {
|
||||
supportingFiles.put(entry.getKey(), assets);
|
||||
for (NamedAsset asset : files.getAssets()) {
|
||||
Resource resource = toResource(asset.getFile());
|
||||
if (resource != null) {
|
||||
supportingFiles
|
||||
.computeIfAbsent(asset.getKey(), key -> new ArrayList<>())
|
||||
.add(resource);
|
||||
}
|
||||
}
|
||||
return new PolicyInputs(primary, supportingFiles);
|
||||
}
|
||||
|
||||
/**
|
||||
* A progress listener that forwards each step transition to the SSE stream as a "step" event.
|
||||
*/
|
||||
private PolicyProgressListener streamListener(SseEmitter emitter) {
|
||||
return new PolicyProgressListener() {
|
||||
@Override
|
||||
@@ -320,8 +334,8 @@ public class PolicyController {
|
||||
try {
|
||||
emitter.send(SseEmitter.event().name(name).data(data, MediaType.APPLICATION_JSON));
|
||||
} catch (IOException | IllegalStateException e) {
|
||||
// Client disconnected or the emitter already closed. The run continues and its results
|
||||
// remain downloadable via the job endpoints; nothing useful left to stream.
|
||||
// Client gone or emitter closed. The run continues and outputs stay downloadable via
|
||||
// the job endpoints.
|
||||
log.debug("Dropping policy SSE event '{}': {}", name, e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -332,20 +346,27 @@ public class PolicyController {
|
||||
return resources;
|
||||
}
|
||||
for (MultipartFile file : files) {
|
||||
Resource resource = toResource(file);
|
||||
if (resource != null) {
|
||||
resources.add(resource);
|
||||
}
|
||||
}
|
||||
return resources;
|
||||
}
|
||||
|
||||
/** Spool a single uploaded file to a managed temp file, preserving its name; null if empty. */
|
||||
private Resource toResource(MultipartFile file) throws IOException {
|
||||
if (file == null || file.isEmpty()) {
|
||||
continue;
|
||||
return null;
|
||||
}
|
||||
TempFile tempFile = tempFileManager.createManagedTempFile("policy-run");
|
||||
file.transferTo(tempFile.getPath());
|
||||
final String originalName = Filenames.toSimpleFileName(file.getOriginalFilename());
|
||||
resources.add(
|
||||
new FileSystemResource(tempFile.getFile()) {
|
||||
return new FileSystemResource(tempFile.getFile()) {
|
||||
@Override
|
||||
public String getFilename() {
|
||||
return originalName;
|
||||
}
|
||||
});
|
||||
}
|
||||
return resources;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
package stirling.software.proprietary.policy.controller;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
|
||||
import jakarta.validation.Valid;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* The files supplied to a policy run: the primary documents and any keyed supporting assets. Bound
|
||||
* from the multipart request via {@code @ModelAttribute}; the pipeline definition itself travels as
|
||||
* a separate typed {@code json} part.
|
||||
*
|
||||
* <p>Wire form: {@code fileInput} (repeated) for primaries, and {@code assets[i].key} / {@code
|
||||
* assets[i].file} for each supporting asset.
|
||||
*/
|
||||
@Data
|
||||
@Schema(description = "Files for a policy run: primary documents plus keyed supporting assets")
|
||||
public class PolicyRunFiles {
|
||||
|
||||
@Schema(description = "Primary input documents", format = "binary")
|
||||
private List<MultipartFile> fileInput = new ArrayList<>();
|
||||
|
||||
@Valid
|
||||
@Schema(description = "Supporting files, each bound to the asset key its step references")
|
||||
private List<NamedAsset> assets = new ArrayList<>();
|
||||
}
|
||||
+26
-43
@@ -33,31 +33,24 @@ import stirling.software.proprietary.policy.output.PolicyOutputSink;
|
||||
import stirling.software.proprietary.policy.progress.PolicyProgressListener;
|
||||
|
||||
/**
|
||||
* Runs pipelines asynchronously as tracked jobs.
|
||||
* Runs pipelines asynchronously as tracked jobs. {@link #submit} returns a run id immediately; the
|
||||
* pipeline runs on a virtual thread (so a step blocked on a slow tool does not hold a platform
|
||||
* thread). Drives {@link PolicyExecutor} for the step loop, projects status/outputs into {@link
|
||||
* TaskManager} (existing job endpoints work unchanged), and keeps live state in {@link
|
||||
* PolicyRunRegistry}.
|
||||
*
|
||||
* <p>Each run is the unit of async work: {@link #submit} returns a run id immediately and the
|
||||
* pipeline executes on a virtual thread, so a step blocking on a slow tool does not tie up a
|
||||
* platform thread. The run drives {@link PolicyExecutor} for the actual step loop, registers its
|
||||
* outputs and progress with {@link TaskManager} (so the existing job status/download endpoints work
|
||||
* unchanged), and keeps rich state in {@link PolicyRunRegistry}.
|
||||
*
|
||||
* <p>The engine deliberately manages its own virtual-thread execution rather than routing through
|
||||
* {@code JobExecutorService}: that path force-completes a job once its work returns, which is
|
||||
* incompatible with a run that suspends in {@code WAITING_FOR_INPUT}. It still applies the shared
|
||||
* {@link ResourceMonitor}/{@link JobQueue} admission control, so heavy runs queue under load
|
||||
* instead of oversubscribing.
|
||||
* <p>Manages its own virtual-thread execution rather than {@code JobExecutorService}, which
|
||||
* force-completes a job once its work returns: incompatible with a run that suspends in {@code
|
||||
* WAITING_FOR_INPUT}. Still applies the shared {@link ResourceMonitor}/{@link JobQueue} admission
|
||||
* control so heavy runs queue under load.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class PolicyEngine {
|
||||
|
||||
/**
|
||||
* Resource weight of a pipeline run for admission control. A run chains many tools and holds
|
||||
* intermediate files, so it is weighted as heavy work: the shared {@link ResourceMonitor}
|
||||
* should let it start while the system is healthy but hold it back under memory/CPU pressure.
|
||||
* See {@link ResourceMonitor#shouldQueueJob(int)} for how a weight maps to that decision.
|
||||
*/
|
||||
// Admission weight for one run. Weighted heavy: a run chains many tools and holds intermediate
|
||||
// files. See ResourceMonitor#shouldQueueJob(int).
|
||||
private static final int RUN_RESOURCE_WEIGHT = 50;
|
||||
|
||||
private final PolicyExecutor stepExecutor;
|
||||
@@ -72,16 +65,14 @@ public class PolicyEngine {
|
||||
private final ExecutorService asyncExecutor = ExecutorFactory.newVirtualThreadExecutor();
|
||||
|
||||
/**
|
||||
* Submit a pipeline to run asynchronously. The returned handle's run id scopes a job in {@link
|
||||
* TaskManager}, so progress (notes), status, and result files are observable via the existing
|
||||
* job endpoints as well as via {@link #getRun(String)}; its completion future resolves when the
|
||||
* run reaches a terminal or paused state.
|
||||
* Submit a pipeline to run asynchronously. The handle's run id scopes a {@link TaskManager} job
|
||||
* (status/notes/results observable via the job endpoints); its future resolves when the run
|
||||
* reaches a terminal or paused state.
|
||||
*/
|
||||
public PolicyRunHandle submit(
|
||||
PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) {
|
||||
// Scope the run id to the current user (on this request thread) so the file-download
|
||||
// ownership check passes; NoOpJobOwnershipService returns the id unchanged when security
|
||||
// is off.
|
||||
// Scope the run id to the current user (this request thread) so the file-download
|
||||
// ownership check passes. No-op when security is off.
|
||||
String runId = jobOwnershipService.createScopedJobKey(UUID.randomUUID().toString());
|
||||
taskManager.createTask(runId);
|
||||
PolicyRun run = new PolicyRun(runId, definition);
|
||||
@@ -90,9 +81,8 @@ public class PolicyEngine {
|
||||
PolicyProgressListener tracking = trackingListener(runId, run, listener);
|
||||
Runnable task = () -> runToCompletion(run, inputs, tracking, completion);
|
||||
|
||||
// Each run is one admission unit; steps run synchronously within it, so this gates heavy
|
||||
// work under load without the pool-within-pool risk of queueing each tool call. Under
|
||||
// resource pressure the run waits in the shared JobQueue; otherwise it starts immediately.
|
||||
// One admission unit per run; steps run synchronously within it, so this gates heavy work
|
||||
// without the pool-within-pool risk of queueing each tool call.
|
||||
if (resourceMonitor.shouldQueueJob(RUN_RESOURCE_WEIGHT)) {
|
||||
log.debug("Queueing policy run {} under resource pressure", runId);
|
||||
jobQueue.queueJob(
|
||||
@@ -110,10 +100,7 @@ public class PolicyEngine {
|
||||
return new PolicyRunHandle(runId, completion);
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a stored policy on demand. Builds the policy's pipeline and submits it. {@code enabled}
|
||||
* gates automatic triggering, not explicit runs, so this runs regardless of that flag.
|
||||
*/
|
||||
/** Run a stored policy on demand. {@code enabled} gates triggers, not explicit runs. */
|
||||
public PolicyRunHandle runPolicy(
|
||||
Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
|
||||
return submit(policy.toDefinition(), inputs, listener);
|
||||
@@ -124,8 +111,7 @@ public class PolicyEngine {
|
||||
}
|
||||
|
||||
/**
|
||||
* Request cancellation of a run. Stage 1 marks the run cancelled in the registry if it has not
|
||||
* already finished; interrupting an in-flight tool call lands in a later stage.
|
||||
* Mark a run cancelled if not already finished. Does not yet interrupt an in-flight tool call.
|
||||
*/
|
||||
public boolean cancel(String runId) {
|
||||
PolicyRun run = registry.get(runId);
|
||||
@@ -139,10 +125,7 @@ public class PolicyEngine {
|
||||
return cancelled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume a run paused in {@code WAITING_FOR_INPUT}. Not yet implemented; the run shape and
|
||||
* {@link WaitState} snapshot are in place so this can be added without reworking the engine.
|
||||
*/
|
||||
/** Resume a run paused in {@code WAITING_FOR_INPUT}. Not yet implemented. */
|
||||
public String resume(String runId, List<Resource> additionalInputs) {
|
||||
throw new UnsupportedOperationException("Pause/resume is not yet implemented");
|
||||
}
|
||||
@@ -163,8 +146,8 @@ public class PolicyEngine {
|
||||
taskManager.setComplete(runId);
|
||||
run.complete(outputs);
|
||||
} catch (PolicyInputRequiredException e) {
|
||||
// Designed-for path: suspend the run rather than fail it. Persist intermediates as
|
||||
// fileIds so the run can resume after this worker thread is gone.
|
||||
// Expected path: suspend rather than fail. Persist intermediates as fileIds so the run
|
||||
// can resume after this worker thread is gone.
|
||||
WaitState wait = suspend(e);
|
||||
run.waitForInput(wait);
|
||||
taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
|
||||
@@ -183,15 +166,15 @@ public class PolicyEngine {
|
||||
run.fail(message);
|
||||
taskManager.setError(runId, message);
|
||||
} finally {
|
||||
// Always resolve the handle with the run's final state so stream/await callers unblock.
|
||||
// Always resolve so stream/await callers unblock.
|
||||
completion.complete(run);
|
||||
}
|
||||
}
|
||||
|
||||
private ResponseEntity<?> failRejectedRun(
|
||||
PolicyRun run, CompletableFuture<PolicyRun> completion, Throwable ex) {
|
||||
// Only reached if the run never started (e.g. the queue was full). A run that started
|
||||
// always resolves its own completion in runToCompletion.
|
||||
// Only reached if the run never started (e.g. queue full); a started run resolves its own
|
||||
// completion in runToCompletion.
|
||||
if (!completion.isDone()) {
|
||||
String message = "Policy run could not be queued: " + ex.getMessage();
|
||||
log.error("Policy run {} was not admitted: {}", run.getRunId(), ex.getMessage());
|
||||
|
||||
+3
-6
@@ -7,11 +7,8 @@ import org.springframework.core.io.Resource;
|
||||
import tools.jackson.databind.JsonNode;
|
||||
|
||||
/**
|
||||
* Result of running a pipeline through {@link PolicyExecutor}.
|
||||
*
|
||||
* <p>{@code files} are the final output resources (temp files, not yet stored to {@code
|
||||
* FileStorage}). {@code report} is the structured metadata payload captured from the last step that
|
||||
* produced one (a JSON body, or an {@code X-Stirling-Tool-Report} header), with {@code reportTool}
|
||||
* naming the step it came from; both are null when no step produced a report.
|
||||
* Result of a {@link PolicyExecutor} run. {@code files} are final temp files (not yet stored).
|
||||
* {@code report}/{@code reportTool} carry the last step's structured report and its operation, or
|
||||
* null if no step produced one.
|
||||
*/
|
||||
public record PolicyExecutionResult(List<Resource> files, JsonNode report, String reportTool) {}
|
||||
|
||||
+27
-55
@@ -35,15 +35,11 @@ import tools.jackson.databind.JsonNode;
|
||||
import tools.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Runs an ordered chain of tool steps, chaining each step's output files into the next step's
|
||||
* input.
|
||||
* Runs an ordered chain of tool steps, feeding each step's output files into the next.
|
||||
*
|
||||
* <p>This is the single execution loop for the proprietary surface (AI plans now;
|
||||
* manually-triggered runs and watched folders later). Each step is dispatched synchronously via
|
||||
* {@link InternalApiClient} loopback HTTP: the tool runs in its own handler and returns its file
|
||||
* inline. The caller decides how to run the executor itself (the AI turn loop calls it directly;
|
||||
* the engine runs it on a virtual thread for async runs). Files cross step boundaries as {@link
|
||||
* Resource} temp files; they are only persisted to durable storage at the run boundaries by the
|
||||
* <p>Steps dispatch synchronously via {@link InternalApiClient} loopback HTTP (each tool runs in
|
||||
* its own handler, returns its file inline). The caller controls threading. Files cross step
|
||||
* boundaries as {@link Resource} temp files and are only persisted at the run boundaries by the
|
||||
* caller.
|
||||
*/
|
||||
@Slf4j
|
||||
@@ -58,25 +54,16 @@ public class PolicyExecutor {
|
||||
private final TempFileManager tempFileManager;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
/**
|
||||
* Internal value-class for tool responses. {@code files} holds any result files (typically one;
|
||||
* multiple for ZIP-response tools). {@code report} holds an optional structured metadata
|
||||
* payload the tool chose to surface alongside (or instead of) a file.
|
||||
*/
|
||||
// files: result files (one, or many for ZIP-response tools). report: optional structured
|
||||
// payload the tool surfaced alongside or instead of a file.
|
||||
private record ToolResult(List<Resource> files, JsonNode report) {}
|
||||
|
||||
/**
|
||||
* Execute every step in {@code definition} in order, feeding each step's output into the next.
|
||||
* Supporting files supplied in {@code inputs} are bound to steps' named file fields and never
|
||||
* enter the document stream.
|
||||
* Run every step in order, feeding each step's output into the next. Supporting files in {@code
|
||||
* inputs} bind to named file fields and never enter the document stream.
|
||||
*
|
||||
* @param definition the pipeline to run (must have at least one step)
|
||||
* @param inputs the primary documents plus the named supporting-file store
|
||||
* @param listener receives per-step progress
|
||||
* @return the final output files plus the last structured report produced, if any
|
||||
* @throws InternalApiTimeoutException if a tool does not respond within its read timeout
|
||||
* @throws IOException if a tool returns a non-OK response, references a missing supporting
|
||||
* file, or a file cannot be read
|
||||
* @throws IOException on a non-OK tool response, a missing supporting file, or a read failure
|
||||
*/
|
||||
public PolicyExecutionResult execute(
|
||||
PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener)
|
||||
@@ -88,7 +75,7 @@ public class PolicyExecutor {
|
||||
|
||||
List<Resource> currentFiles = inputs.primary();
|
||||
Map<String, List<Resource>> supportingFiles = inputs.supportingFiles();
|
||||
// Propagate the *last* non-null report; the terminal step defines the output.
|
||||
// Last non-null report wins: the terminal step defines the output.
|
||||
JsonNode lastReport = null;
|
||||
String lastReportTool = null;
|
||||
|
||||
@@ -113,13 +100,9 @@ public class PolicyExecutor {
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a single tool step. If the endpoint accepts multiple files, all files are sent in one
|
||||
* call. Otherwise, the endpoint is called once per file. ZIP responses are unpacked so each
|
||||
* inner file is treated as its own result (e.g. split outputs a ZIP of pages).
|
||||
*
|
||||
* <p>A structured {@code report} may be returned alongside (or instead of) files; see {@link
|
||||
* ToolResult}. For per-file dispatch (single-input endpoints called once per input), the first
|
||||
* non-null report wins.
|
||||
* Multi-input endpoints get all files in one call; others are called once per file. ZIP
|
||||
* responses are unpacked so each inner file is its own result (e.g. split). For per-file
|
||||
* dispatch the first non-null report wins.
|
||||
*/
|
||||
private ToolResult executeStep(
|
||||
PipelineStep step,
|
||||
@@ -146,17 +129,10 @@ public class PolicyExecutor {
|
||||
}
|
||||
|
||||
/**
|
||||
* Call an endpoint and return its result files and optional report.
|
||||
*
|
||||
* <ul>
|
||||
* <li>JSON body (Content-Type: application/json): the entire body is the report, no files are
|
||||
* returned.
|
||||
* <li>File body (PDF etc.): the file is returned; if an {@link
|
||||
* AiToolResponseHeaders#TOOL_REPORT} header is present, its (minified JSON) value is
|
||||
* parsed as the report.
|
||||
* <li>ZIP responses declared by the tool metadata service are unpacked so callers always see
|
||||
* a flat list of result files.
|
||||
* </ul>
|
||||
* Call an endpoint, returning result files and optional report. Response handling: JSON body is
|
||||
* the report with no file; a file body returns the file plus any {@link
|
||||
* AiToolResponseHeaders#TOOL_REPORT} header report; ZIP responses (per tool metadata) are
|
||||
* unpacked to a flat file list.
|
||||
*/
|
||||
private ToolResult callEndpoint(
|
||||
PipelineStep step, List<Resource> files, Map<String, List<Resource>> supportingFiles)
|
||||
@@ -166,8 +142,8 @@ public class PolicyExecutor {
|
||||
for (Resource file : files) {
|
||||
body.add("fileInput", file);
|
||||
}
|
||||
// Bind supporting files to their named tool fields (e.g. stampImage, overlayFiles). These
|
||||
// come from the run's named asset store, not the document stream.
|
||||
// Bind supporting files to named tool fields (e.g. stampImage); from the asset store, not
|
||||
// the document stream.
|
||||
for (Map.Entry<String, String> binding : step.fileParameters().entrySet()) {
|
||||
String fieldName = binding.getKey();
|
||||
String assetKey = binding.getValue();
|
||||
@@ -189,9 +165,9 @@ public class PolicyExecutor {
|
||||
for (Map.Entry<String, Object> entry : step.parameters().entrySet()) {
|
||||
if (entry.getValue() instanceof List<?> list) {
|
||||
if (containsStructuredElements(list)) {
|
||||
// Endpoints binding lists of structured objects (e.g. /security/redact's
|
||||
// redactions, /general/edit-text's edits) parse a single JSON string field via
|
||||
// a property editor. Pre-serialize the whole list so binding succeeds.
|
||||
// These endpoints (e.g. /security/redact redactions, /general/edit-text edits)
|
||||
// bind a list of structured objects from a single JSON string field via a
|
||||
// property editor, so pre-serialize the whole list.
|
||||
body.add(entry.getKey(), objectMapper.writeValueAsString(list));
|
||||
} else {
|
||||
for (Object item : list) {
|
||||
@@ -209,8 +185,8 @@ public class PolicyExecutor {
|
||||
}
|
||||
Resource resource = response.getBody();
|
||||
|
||||
// Filter operations return an empty body to signal the file was filtered out: drop it
|
||||
// rather than forwarding a zero-byte document.
|
||||
// Filter ops return an empty body to mean "filtered out": drop it rather than forward a
|
||||
// zero-byte document.
|
||||
if (isFilterOperation(endpointPath) && isEmpty(resource)) {
|
||||
return new ToolResult(List.of(), null);
|
||||
}
|
||||
@@ -218,7 +194,7 @@ public class PolicyExecutor {
|
||||
HttpHeaders headers = response.getHeaders();
|
||||
MediaType contentType = headers.getContentType();
|
||||
|
||||
// JSON-only response: the whole body is the structured report, no result file.
|
||||
// JSON-only response: whole body is the report, no file.
|
||||
if (contentType != null && MediaType.APPLICATION_JSON.isCompatibleWith(contentType)) {
|
||||
try (InputStream is = resource.getInputStream()) {
|
||||
JsonNode report = objectMapper.readTree(is);
|
||||
@@ -233,10 +209,7 @@ public class PolicyExecutor {
|
||||
return new ToolResult(List.of(resource), report);
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse the optional {@link AiToolResponseHeaders#TOOL_REPORT} header into a {@link JsonNode},
|
||||
* or return null.
|
||||
*/
|
||||
/** Parse the optional {@link AiToolResponseHeaders#TOOL_REPORT} header, or null. */
|
||||
private JsonNode parseReportHeader(HttpHeaders headers, String endpointPath) {
|
||||
String raw = headers.getFirst(AiToolResponseHeaders.TOOL_REPORT);
|
||||
if (raw == null || raw.isBlank()) {
|
||||
@@ -264,8 +237,7 @@ public class PolicyExecutor {
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail the run if any document in the primary stream is not a file type the step accepts. An
|
||||
* endpoint that declares no specific input type accepts anything.
|
||||
* Fail if any primary-stream file is a type the step rejects. No declared type means anything.
|
||||
*/
|
||||
private void requireAcceptedTypes(String operation, List<Resource> files) throws IOException {
|
||||
List<String> accepted = toolMetadataService.getExtensionTypes(false, operation);
|
||||
|
||||
+3
-9
@@ -7,15 +7,9 @@ import org.springframework.core.io.Resource;
|
||||
import lombok.Getter;
|
||||
|
||||
/**
|
||||
* Thrown by a step to signal that the run cannot proceed without further user input, pausing the
|
||||
* run in {@code WAITING_FOR_INPUT} rather than failing it.
|
||||
*
|
||||
* <p>Carries everything needed to resume: a human-readable reason, the 0-based index of the step to
|
||||
* resume from, and the intermediate files produced so far. The engine persists those files and
|
||||
* suspends the run.
|
||||
*
|
||||
* <p>Defined now to fix the run shape; no step throws it yet, and the resume handshake is
|
||||
* implemented in a later stage.
|
||||
* Thrown by a step that needs further user input, pausing the run in {@code WAITING_FOR_INPUT}
|
||||
* instead of failing. Carries the resume reason, 0-based resume step index, and intermediate files;
|
||||
* the engine persists those and suspends. Not yet thrown by any step.
|
||||
*/
|
||||
@Getter
|
||||
public class PolicyInputRequiredException extends RuntimeException {
|
||||
|
||||
+4
-7
@@ -5,12 +5,9 @@ import java.util.concurrent.CompletableFuture;
|
||||
import stirling.software.proprietary.policy.model.PolicyRun;
|
||||
|
||||
/**
|
||||
* Returned by {@link PolicyEngine#submit}: the run id (for status polling and result download) plus
|
||||
* a future that resolves when the run reaches a terminal or paused state.
|
||||
*
|
||||
* <p>The completion future lets callers react to the end of a run (e.g. an SSE endpoint sending a
|
||||
* final event and closing the stream) without polling. It carries the {@link PolicyRun} whose
|
||||
* status describes the outcome (completed, failed, cancelled, or waiting for input); it does not
|
||||
* complete exceptionally for ordinary run failures.
|
||||
* Returned by {@link PolicyEngine#submit}: the run id (status polling, result download) plus a
|
||||
* future that resolves when the run reaches a terminal or paused state. The future carries the
|
||||
* {@link PolicyRun} whose status describes the outcome; it does not complete exceptionally for
|
||||
* ordinary run failures.
|
||||
*/
|
||||
public record PolicyRunHandle(String runId, CompletableFuture<PolicyRun> completion) {}
|
||||
|
||||
+8
-13
@@ -19,16 +19,12 @@ import stirling.software.common.model.ApplicationProperties;
|
||||
import stirling.software.proprietary.policy.model.PolicyRun;
|
||||
|
||||
/**
|
||||
* In-memory store of live {@link PolicyRun} state, keyed by runId. Holds the authoritative run
|
||||
* state machine; durable status/files for download are projected separately into {@code
|
||||
* TaskManager}.
|
||||
* In-memory store of live {@link PolicyRun} state, keyed by runId. Authoritative run state machine;
|
||||
* durable status/files are projected separately into {@code TaskManager}.
|
||||
*
|
||||
* <p>Finished runs are evicted on a fixed interval once they age past {@code
|
||||
* policies.runExpiryMinutes}, mirroring the job-result expiry in {@code TaskManager} so a run's
|
||||
* rich in-memory state does not outlive the process. Only terminal runs are evicted; active and
|
||||
* paused ({@code WAITING_FOR_INPUT}) runs are retained regardless of age. Result files are not
|
||||
* touched here: a run shares its runId with a {@code TaskManager} job, which owns file-lifecycle
|
||||
* cleanup, so eviction only frees this map's entry.
|
||||
* <p>A scheduled sweep evicts only terminal runs aged past {@code policies.runExpiryMinutes};
|
||||
* active and paused runs are kept regardless of age. Eviction frees only this map's entry: the
|
||||
* shared {@code TaskManager} job owns file-lifecycle cleanup.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -61,7 +57,7 @@ public class PolicyRunRegistry {
|
||||
return runs.values();
|
||||
}
|
||||
|
||||
/** Scheduled hook: evict terminal runs that finished before the expiry window. */
|
||||
/** Scheduled sweep entry point. */
|
||||
private void evictExpiredRuns() {
|
||||
try {
|
||||
evictExpired(Instant.now().minus(runExpiry));
|
||||
@@ -71,9 +67,8 @@ public class PolicyRunRegistry {
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove every terminal run last updated before {@code cutoff}; active and paused runs are kept
|
||||
* regardless of age. Returns the number evicted. Package-visible so the scheduled sweep and
|
||||
* tests exercise the same path with an explicit cutoff.
|
||||
* Evict terminal runs last updated before {@code cutoff}, returning the count. Package-visible
|
||||
* so the sweep and tests share one path with an explicit cutoff.
|
||||
*/
|
||||
int evictExpired(Instant cutoff) {
|
||||
int removed = 0;
|
||||
|
||||
+6
-17
@@ -20,14 +20,8 @@ import stirling.software.proprietary.policy.model.PolicyRunStatus;
|
||||
import stirling.software.proprietary.policy.progress.PolicyProgressListener;
|
||||
|
||||
/**
|
||||
* Runs policies, and is the one place that knows how to turn a policy's configured {@link InputSpec
|
||||
* sources} into actual runs. Triggers (schedule, and future webhook/folder-watch) decide
|
||||
* <em>when</em> to run and call {@link #run(Policy)}; they never touch sources themselves. The
|
||||
* controller uses the supplied-input and ad-hoc entry points for on-demand work.
|
||||
*
|
||||
* <p>This is the seam that keeps triggers and sources independent: a trigger depends on the runner,
|
||||
* the runner depends on the {@link InputSource} beans, and a source depends on neither - it just
|
||||
* yields {@link ResolvedInput units of work}, each carrying its own completion hook.
|
||||
* Turns a policy's configured {@link InputSpec sources} into runs. Triggers decide <em>when</em>
|
||||
* and call {@link #run(Policy)}; the controller uses the supplied-input and ad-hoc entry points.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -38,10 +32,9 @@ public class PolicyRunner {
|
||||
private final List<InputSource> inputSources;
|
||||
|
||||
/**
|
||||
* Run a policy by pulling from every source it configures: each source yields zero or more
|
||||
* units of work, and each unit becomes its own run so one failure does not affect the others. A
|
||||
* policy with no sources runs once with no input files (a generator pipeline). Used by
|
||||
* automatic triggers.
|
||||
* Trigger entry point. Pulls every configured source; each yielded unit becomes its own run so
|
||||
* one failure does not affect the others. No sources means one run with no input (generator
|
||||
* pipeline).
|
||||
*/
|
||||
public void run(Policy policy) {
|
||||
List<InputSpec> sources = policy.sources();
|
||||
@@ -54,11 +47,7 @@ public class PolicyRunner {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a stored policy on files supplied directly by the caller (e.g. a manual run with
|
||||
* uploads), bypassing its configured sources. Returns the run handle so callers can stream
|
||||
* progress.
|
||||
*/
|
||||
/** Run a stored policy on caller-supplied files (e.g. manual upload), bypassing its sources. */
|
||||
public PolicyRunHandle runWith(
|
||||
Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
|
||||
return policyEngine.runPolicy(policy, inputs, listener);
|
||||
|
||||
+4
-8
@@ -15,12 +15,9 @@ import stirling.software.proprietary.policy.output.PolicyOutputSink;
|
||||
import stirling.software.proprietary.policy.trigger.PolicyTrigger;
|
||||
|
||||
/**
|
||||
* Validates a policy's trigger, sources, and output configuration by delegating each facet to the
|
||||
* bean that handles its type. Called when a policy is saved so a misconfigured schedule, missing
|
||||
* folder directory, or unknown type fails fast instead of silently misbehaving at run time.
|
||||
*
|
||||
* <p>The trigger is optional (a {@code null} trigger is a manual-only policy and needs no
|
||||
* validation); every configured source is validated.
|
||||
* Validates a policy at save time by delegating each facet (trigger, sources, output) to the bean
|
||||
* that handles its type, so a misconfiguration fails fast rather than at run time. A null trigger
|
||||
* is a manual-only policy and skips trigger validation.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@@ -31,8 +28,7 @@ public class PolicyValidator {
|
||||
private final List<PolicyOutputSink> outputSinks;
|
||||
|
||||
/**
|
||||
* @throws IllegalArgumentException if any facet's type is unknown or its configuration is
|
||||
* invalid
|
||||
* @throws IllegalArgumentException if any facet's type is unknown or its config is invalid
|
||||
*/
|
||||
public void validate(Policy policy) {
|
||||
if (policy.trigger() != null) {
|
||||
|
||||
+8
-17
@@ -22,21 +22,13 @@ import stirling.software.proprietary.policy.model.InputSpec;
|
||||
import stirling.software.proprietary.policy.model.PolicyInputs;
|
||||
|
||||
/**
|
||||
* Reads input files from a directory. Each ready file becomes its own unit of work (one run per
|
||||
* file) so a failure on one file does not affect the others.
|
||||
* Reads input files from a directory; each ready file is its own unit of work so one failure does
|
||||
* not affect the others.
|
||||
*
|
||||
* <p>Two modes via the {@code mode} option:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code "consume"} (default) - claim each file by moving it into {@code
|
||||
* .stirling/processing}, then route it to {@code .stirling/done} or {@code .stirling/error}
|
||||
* when its run finishes. Each file is processed once; right for "process new arrivals" (and
|
||||
* the basis of watched folders).
|
||||
* <li>{@code "snapshot"} - read the directory's current files without moving them; every run sees
|
||||
* the full set again. Right for "always regenerate from a fixed input set".
|
||||
* </ul>
|
||||
*
|
||||
* Readiness is checked first (via {@link FileReadinessChecker}) so files mid-write are skipped.
|
||||
* <p>Mode option: "consume" (default) claims each file by moving it into {@code
|
||||
* .stirling/processing} then routes it to {@code .stirling/done} or {@code .stirling/error}, so
|
||||
* each file runs once; "snapshot" reads without moving, so every run sees the full set. Readiness
|
||||
* is checked first so files mid-write are skipped.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -44,7 +36,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs;
|
||||
public class FolderInputSource implements InputSource {
|
||||
|
||||
private static final String TYPE = FolderAccessGuard.FOLDER_TYPE;
|
||||
// Bookkeeping lives under one hidden namespace dir so the watched folder stays tidy.
|
||||
// Bookkeeping lives under one hidden dir so the watched folder stays tidy.
|
||||
private static final String WORK_SUBDIR = ".stirling";
|
||||
private static final String PROCESSING_SUBDIR = "processing";
|
||||
private static final String DONE_SUBDIR = "done";
|
||||
@@ -107,6 +99,7 @@ public class FolderInputSource implements InputSource {
|
||||
return work;
|
||||
}
|
||||
|
||||
// Atomic move into processing/: only one sweep can win the claim, the rest see the file gone.
|
||||
private Path claim(Path inputDir, Path file) {
|
||||
try {
|
||||
Path processingDir = workDir(inputDir, PROCESSING_SUBDIR);
|
||||
@@ -135,7 +128,6 @@ public class FolderInputSource implements InputSource {
|
||||
}
|
||||
}
|
||||
|
||||
/** A bookkeeping subdirectory under the watched folder's {@code .stirling} namespace. */
|
||||
private static Path workDir(Path inputDir, String subdir) {
|
||||
return inputDir.resolve(WORK_SUBDIR).resolve(subdir);
|
||||
}
|
||||
@@ -166,7 +158,6 @@ public class FolderInputSource implements InputSource {
|
||||
}
|
||||
}
|
||||
|
||||
/** The typed, validated form of a folder source's options: the directory and dedup mode. */
|
||||
record FolderConfig(Path directory, boolean snapshot) {
|
||||
|
||||
private static final String DIRECTORY_OPTION = "directory";
|
||||
|
||||
+8
-19
@@ -7,14 +7,9 @@ import java.util.List;
|
||||
import stirling.software.proprietary.policy.model.InputSpec;
|
||||
|
||||
/**
|
||||
* Resolves one of a policy's {@link InputSpec sources} into the files to run on - answering
|
||||
* <em>where</em> a run's files come from, independent of <em>when</em> it runs. The counterpart of
|
||||
* {@code PolicyOutputSink}: implementations are beans selected by {@link #supports(InputSpec)}, so
|
||||
* a new source kind (folder, S3) is just a new bean.
|
||||
*
|
||||
* <p>Driven by the {@code PolicyRunner}, which a trigger calls when a policy is due; a source is
|
||||
* passive and knows nothing about what triggered the run. A manual run may instead supply files
|
||||
* directly and bypass sources entirely.
|
||||
* Resolves a policy {@link InputSpec} into the files to run on. Implementations are beans selected
|
||||
* by {@link #supports(InputSpec)}, so a new source kind (folder, S3) is just a new bean. A manual
|
||||
* run may supply files directly and bypass sources entirely.
|
||||
*/
|
||||
public interface InputSource {
|
||||
|
||||
@@ -24,24 +19,18 @@ public interface InputSource {
|
||||
/** Whether this source can handle the given spec. */
|
||||
boolean supports(InputSpec spec);
|
||||
|
||||
/**
|
||||
* Check that an input spec is usable, throwing {@link IllegalArgumentException} if not. Called
|
||||
* when a policy is saved so misconfiguration fails fast rather than at run time.
|
||||
*/
|
||||
/** Throws {@link IllegalArgumentException} on bad config. Called on save to fail fast. */
|
||||
default void validate(InputSpec spec) {}
|
||||
|
||||
/**
|
||||
* Resolve the spec into zero or more units of work, each carrying the files for one run and a
|
||||
* completion hook. Returning an empty list means there is nothing to run right now.
|
||||
* Resolve the spec into zero or more units of work, each carrying one run's files and a
|
||||
* completion hook. Empty list means nothing to run right now.
|
||||
*/
|
||||
List<ResolvedInput> resolve(InputSpec spec) throws IOException;
|
||||
|
||||
/**
|
||||
* The local filesystem directories this source draws from, if any, for triggers that want to
|
||||
* react to changes there (the folder-watch trigger) rather than poll. Advisory only: it merely
|
||||
* tells a trigger <em>where</em> to watch; resolving the spec into files is still this source's
|
||||
* job via {@link #resolve}. Non-filesystem sources (S3, ...) return an empty list and so are
|
||||
* simply not watchable. Default: nothing to watch.
|
||||
* Filesystem dirs this source draws from, for the folder-watch trigger. Advisory: resolving is
|
||||
* still done by {@link #resolve}. Non-filesystem sources return empty and are not watchable.
|
||||
*/
|
||||
default List<Path> watchTargets(InputSpec spec) {
|
||||
return List.of();
|
||||
|
||||
+4
-5
@@ -5,10 +5,9 @@ import java.util.function.Consumer;
|
||||
import stirling.software.proprietary.policy.model.PolicyInputs;
|
||||
|
||||
/**
|
||||
* One unit of work produced by an {@link InputSource}: the files to run plus a completion callback
|
||||
* invoked with the run's success once it finishes (e.g. a folder source routes the input to {@code
|
||||
* .stirling/done} or {@code .stirling/error}). A source may return several of these (e.g. one per
|
||||
* file).
|
||||
* One unit of work from an {@link InputSource}: the files to run plus a completion callback invoked
|
||||
* with the run's success (e.g. a folder source routes the input to done/error). A source may return
|
||||
* several of these, one per file.
|
||||
*/
|
||||
public record ResolvedInput(PolicyInputs inputs, Consumer<Boolean> onComplete) {
|
||||
|
||||
@@ -16,7 +15,7 @@ public record ResolvedInput(PolicyInputs inputs, Consumer<Boolean> onComplete) {
|
||||
onComplete = onComplete == null ? success -> {} : onComplete;
|
||||
}
|
||||
|
||||
/** A unit of work with no completion side effect. */
|
||||
/** No completion side effect. */
|
||||
public static ResolvedInput of(PolicyInputs inputs) {
|
||||
return new ResolvedInput(inputs, success -> {});
|
||||
}
|
||||
|
||||
+2
-9
@@ -3,15 +3,8 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* One source a policy's input files come from. {@code type} selects an {@code InputSource}
|
||||
* ("folder", and in future "s3", ...); {@code options} carries source-specific configuration (a
|
||||
* directory, a bucket, a dedup mode, ...).
|
||||
*
|
||||
* <p>A policy holds a list of these; a run pulls from every one. An empty list means the policy
|
||||
* runs with no input files (a generator pipeline, or files supplied directly to a manual run).
|
||||
*
|
||||
* <p>Data-driven and parallel to {@link OutputSpec}/{@link TriggerConfig}: a new source kind is a
|
||||
* new {@code type} handled by a new {@code InputSource} bean.
|
||||
* One input source for a policy. {@code type} keys an {@code InputSource} bean; a run pulls from
|
||||
* every source.
|
||||
*/
|
||||
public record InputSpec(String type, Map<String, Object> options) {
|
||||
|
||||
|
||||
+2
-8
@@ -2,19 +2,13 @@ package stirling.software.proprietary.policy.model;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Describes where a pipeline run's output files should be delivered. {@code type} selects a {@code
|
||||
* PolicyOutputSink} (e.g. "inline"); {@code options} carries sink-specific configuration.
|
||||
*
|
||||
* <p>New destinations (folder, S3) are added as new sink beans keyed on a new {@code type} without
|
||||
* changing this shape or the engine.
|
||||
*/
|
||||
/** Where a run's outputs are delivered. {@code type} keys a {@code PolicyOutputSink} bean. */
|
||||
public record OutputSpec(String type, Map<String, Object> options) {
|
||||
public OutputSpec {
|
||||
options = options == null ? Map.of() : options;
|
||||
}
|
||||
|
||||
/** The default destination: store outputs and return them to the caller for download. */
|
||||
/** Default sink: store outputs and return them to the caller for download. */
|
||||
public static OutputSpec inline() {
|
||||
return new OutputSpec("inline", Map.of());
|
||||
}
|
||||
|
||||
+3
-4
@@ -3,11 +3,10 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* An ordered chain of tool steps plus where the output should go.
|
||||
* An ordered chain of tool steps plus an output destination; the unit the engine executes.
|
||||
*
|
||||
* <p>This is the single shape executed by the policy engine, shared by AI plans, manually-triggered
|
||||
* runs, and (later) watched folders. {@code output} may be null for callers that handle result
|
||||
* files themselves (e.g. the AI workflow, which builds its own response payload).
|
||||
* <p>{@code output} may be null for callers that handle result files themselves (e.g. the AI
|
||||
* workflow, which builds its own response payload).
|
||||
*/
|
||||
public record PipelineDefinition(String name, List<PipelineStep> steps, OutputSpec output) {
|
||||
public PipelineDefinition {
|
||||
|
||||
+6
-10
@@ -3,17 +3,13 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* A single tool invocation in a pipeline: the API endpoint path to call and the inputs to pass.
|
||||
* A single tool invocation. {@code operation} is a Stirling endpoint path (e.g. {@code
|
||||
* /api/v1/misc/compress-pdf}) per the {@code InternalApiClient} convention; {@code parameters} are
|
||||
* scalar form fields.
|
||||
*
|
||||
* <p>{@code operation} is a Stirling tool endpoint path (e.g. {@code /api/v1/misc/compress-pdf}),
|
||||
* matching the dispatch convention used by {@code InternalApiClient}. {@code parameters} are the
|
||||
* tool-specific scalar form fields.
|
||||
*
|
||||
* <p>{@code fileParameters} binds a tool's named file fields (beyond the primary {@code fileInput}
|
||||
* stream) to supporting files supplied with the run: it maps the form field name (e.g. {@code
|
||||
* stampImage}, {@code overlayFiles}) to an asset key in the run's supporting-file store. This keeps
|
||||
* supporting inputs (a stamp image, a certificate, an overlay) out of the document stream that
|
||||
* flows step to step.
|
||||
* <p>{@code fileParameters} maps a tool's named file field (e.g. {@code stampImage}, beyond the
|
||||
* primary {@code fileInput} stream) to an asset key in the run's supporting-file store, keeping
|
||||
* supporting inputs out of the document stream that flows step to step.
|
||||
*/
|
||||
public record PipelineStep(
|
||||
String operation, Map<String, Object> parameters, Map<String, String> fileParameters) {
|
||||
|
||||
+5
-10
@@ -3,16 +3,11 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* A stored automation: an ordered chain of tool steps, the sources its input files come from, and
|
||||
* an output destination for the results.
|
||||
* A stored automation: ordered tool steps, input sources, and an output destination.
|
||||
*
|
||||
* <p>Every policy can always be run on demand (manually). It may additionally carry one automatic
|
||||
* {@link TriggerConfig} - usually a schedule - that fires it without a person asking; a {@code
|
||||
* null} trigger means manual-only. A trigger decides <em>when</em> a run happens and a {@link
|
||||
* InputSpec source} decides <em>where</em> its files come from; the two are independent, and a run
|
||||
* pulls from every configured source.
|
||||
*
|
||||
* <p>This is the feature's central configuration object - what a user defines and the engine runs.
|
||||
* <p>Always runnable on demand. An optional {@link TriggerConfig} fires it automatically; a {@code
|
||||
* null} trigger means manual-only. Trigger decides when, {@link InputSpec sources} decide where
|
||||
* files come from; a run pulls from every source.
|
||||
*/
|
||||
public record Policy(
|
||||
String id,
|
||||
@@ -42,7 +37,7 @@ public record Policy(
|
||||
this(id, name, owner, enabled, trigger, List.of(), steps, output);
|
||||
}
|
||||
|
||||
/** The engine-level, trigger-agnostic view of this policy's pipeline. */
|
||||
/** This policy's pipeline as the engine sees it. */
|
||||
public PipelineDefinition toDefinition() {
|
||||
return new PipelineDefinition(name, steps, output);
|
||||
}
|
||||
|
||||
+3
-11
@@ -6,17 +6,9 @@ import java.util.Map;
|
||||
import org.springframework.core.io.Resource;
|
||||
|
||||
/**
|
||||
* The files a run operates on, split into two roles:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code primary} - the documents that flow through the pipeline, each step's output becoming
|
||||
* the next step's input.
|
||||
* <li>{@code supportingFiles} - a named store of auxiliary files (a stamp image, certificate,
|
||||
* overlay, attachments) that steps bind to their named file fields via {@link
|
||||
* PipelineStep#fileParameters()}. These never enter the document stream.
|
||||
* </ul>
|
||||
*
|
||||
* Asset values are lists so a single key can carry multi-file fields (e.g. attachments).
|
||||
* A run's files. {@code primary} documents flow step to step; {@code supportingFiles} are auxiliary
|
||||
* assets bound by key via {@link PipelineStep#fileParameters()} and never enter the document
|
||||
* stream. Asset values are lists so one key can carry a multi-file field (e.g. attachments).
|
||||
*/
|
||||
public record PolicyInputs(List<Resource> primary, Map<String, List<Resource>> supportingFiles) {
|
||||
|
||||
|
||||
+5
-9
@@ -8,12 +8,10 @@ import lombok.Getter;
|
||||
import stirling.software.common.model.job.ResultFile;
|
||||
|
||||
/**
|
||||
* Live, mutable state of a single pipeline run, held in memory by {@code PolicyRunRegistry}.
|
||||
*
|
||||
* <p>This carries the rich execution state (status, step cursor, wait state) that the job system's
|
||||
* {@code JobResult} does not model. The run is also projected into {@code TaskManager} for
|
||||
* cluster-visible status, progress notes, and file download; this object is the authoritative
|
||||
* source of the state machine.
|
||||
* Live, mutable state of one pipeline run, held in memory by {@code PolicyRunRegistry} and the
|
||||
* authoritative source of the state machine. Carries execution state ({@code JobResult} does not
|
||||
* model status/step cursor/wait state); also projected into {@code TaskManager} for cluster-visible
|
||||
* status and download.
|
||||
*/
|
||||
@Getter
|
||||
public class PolicyRun {
|
||||
@@ -69,9 +67,7 @@ public class PolicyRun {
|
||||
touch();
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark cancelled if the run has not already reached a terminal state. Returns whether it did.
|
||||
*/
|
||||
/** Cancels unless already terminal; returns whether it transitioned. */
|
||||
public synchronized boolean cancel() {
|
||||
if (status.isTerminal()) {
|
||||
return false;
|
||||
|
||||
+2
-5
@@ -1,11 +1,8 @@
|
||||
package stirling.software.proprietary.policy.model;
|
||||
|
||||
/**
|
||||
* Lifecycle states of a {@link PolicyRun}.
|
||||
*
|
||||
* <p>{@code WAITING_FOR_INPUT} is modelled now so the engine and run shape support pausing a run
|
||||
* (e.g. a step that blocks for a human decision) without holding a thread; the resume handshake is
|
||||
* implemented in a later stage.
|
||||
* Lifecycle states of a {@link PolicyRun}. {@code WAITING_FOR_INPUT} models a thread-free pause;
|
||||
* the resume handshake lands in a later stage.
|
||||
*/
|
||||
public enum PolicyRunStatus {
|
||||
PENDING,
|
||||
|
||||
+2
-2
@@ -5,8 +5,8 @@ import java.util.List;
|
||||
import stirling.software.common.model.job.ResultFile;
|
||||
|
||||
/**
|
||||
* Read-only view of a {@link PolicyRun} returned by the status endpoint. Output files are surfaced
|
||||
* as {@link ResultFile} so the caller can download each via {@code GET /api/v1/general/files/{id}}.
|
||||
* Read-only view of a {@link PolicyRun} for the status endpoint. Outputs are {@link ResultFile}s,
|
||||
* downloadable via {@code GET /api/v1/general/files/{id}}.
|
||||
*/
|
||||
public record PolicyRunView(
|
||||
String runId,
|
||||
|
||||
+5
-16
@@ -11,16 +11,9 @@ import com.fasterxml.jackson.annotation.JsonSubTypes;
|
||||
import com.fasterxml.jackson.annotation.JsonTypeInfo;
|
||||
|
||||
/**
|
||||
* When a scheduled policy should fire, expressed as intent ("every day at 02:00") rather than a
|
||||
* cron string. A frontend builds one of these from a friendly picker; the schedule trigger asks it
|
||||
* for the next firing after a given moment.
|
||||
*
|
||||
* <p>Sealed so every cadence is an explicit, self-validating shape and adding a new one forces a
|
||||
* new subtype rather than another string convention to learn. The {@code type} discriminator stays
|
||||
* on the wire so the frontend can switch on it without knowing the Java hierarchy.
|
||||
*
|
||||
* <p>Wall-clock kinds ({@link Daily}, {@link Weekly}, {@link Monthly}) are evaluated in the zone of
|
||||
* the {@code after} argument; {@link Every} is a fixed offset and ignores wall-clock time.
|
||||
* A scheduled policy's firing cadence; {@code type} is the JSON discriminator. Wall-clock kinds
|
||||
* ({@link Daily}, {@link Weekly}, {@link Monthly}) evaluate in the {@code after} argument's zone;
|
||||
* {@link Every} is a fixed offset and ignores wall-clock time.
|
||||
*/
|
||||
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
|
||||
@JsonSubTypes({
|
||||
@@ -42,10 +35,7 @@ public sealed interface Schedule {
|
||||
DAYS
|
||||
}
|
||||
|
||||
/**
|
||||
* A fixed cadence repeating from when the policy was last seen: "every 15 minutes", "every 6
|
||||
* hours". Time of day is irrelevant.
|
||||
*/
|
||||
/** A fixed offset from {@code after}: "every 15 minutes", "every 6 hours". No time of day. */
|
||||
record Every(long count, Unit unit) implements Schedule {
|
||||
public Every {
|
||||
if (count <= 0) {
|
||||
@@ -91,8 +81,7 @@ public sealed interface Schedule {
|
||||
|
||||
@Override
|
||||
public ZonedDateTime nextAfter(ZonedDateTime after) {
|
||||
// The soonest of the next 7 days that lands on a chosen weekday, at the configured
|
||||
// time.
|
||||
// Soonest of the next 7 days landing on a chosen weekday, at the configured time.
|
||||
for (int i = 0; i <= 7; i++) {
|
||||
ZonedDateTime candidate = after.plusDays(i).with(at);
|
||||
if (candidate.isAfter(after) && days.contains(candidate.getDayOfWeek())) {
|
||||
|
||||
+3
-11
@@ -3,17 +3,9 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* A {@link Policy}'s optional automatic trigger - what fires it without a person asking. {@code
|
||||
* type} selects a trigger kind ("schedule", and in future "webhook", "folder-watch", ...); {@code
|
||||
* options} carries type-specific configuration (a {@link Schedule}, a webhook secret, ...).
|
||||
*
|
||||
* <p>Manual running is <em>not</em> a trigger kind: every policy can always be run on demand, so a
|
||||
* policy with no automatic trigger simply has a {@code null} {@code TriggerConfig}. A trigger
|
||||
* answers only "when"; where a run's files come from is a separate concern owned by the policy's
|
||||
* {@link InputSpec sources}.
|
||||
*
|
||||
* <p>Data-driven and parallel to {@link OutputSpec}: new trigger kinds are new {@code type} values
|
||||
* handled by a new trigger bean, with no change to the model.
|
||||
* A {@link Policy}'s automatic trigger; {@code type} keys a trigger bean (e.g. "schedule"). Manual
|
||||
* running is not a trigger kind: a manual-only policy carries a {@code null} {@code TriggerConfig}.
|
||||
* Answers only "when"; file sources are the policy's {@link InputSpec}s.
|
||||
*/
|
||||
public record TriggerConfig(String type, Map<String, Object> options) {
|
||||
|
||||
|
||||
+4
-8
@@ -3,14 +3,10 @@ package stirling.software.proprietary.policy.model;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Captured when a run pauses in {@link PolicyRunStatus#WAITING_FOR_INPUT}. Together with the run's
|
||||
* {@link PipelineDefinition} this is the resumable snapshot: {@code resumeStepIndex} is the 0-based
|
||||
* step to continue from, and {@code pendingFileIds} are the intermediate files (stored in {@code
|
||||
* FileStorage}, so they survive the worker thread ending or a node restart) that become the input
|
||||
* to the resumed run.
|
||||
*
|
||||
* <p>Stored as fileIds rather than in-memory resources by design: a paused run must be resumable
|
||||
* long after its worker thread has gone.
|
||||
* Resumable snapshot captured when a run pauses ({@link PolicyRunStatus#WAITING_FOR_INPUT}). {@code
|
||||
* resumeStepIndex} is the 0-based step to continue from; {@code pendingFileIds} are intermediate
|
||||
* files held in {@code FileStorage} (not in-memory resources) so a pause survives the worker thread
|
||||
* ending or a node restart.
|
||||
*/
|
||||
public record WaitState(String reason, int resumeStepIndex, List<String> pendingFileIds) {
|
||||
public WaitState {
|
||||
|
||||
+6
-13
@@ -22,13 +22,10 @@ import stirling.software.proprietary.policy.config.FolderAccessGuard;
|
||||
import stirling.software.proprietary.policy.model.OutputSpec;
|
||||
|
||||
/**
|
||||
* Writes a run's output files to a directory on disk. The destination is the {@code directory}
|
||||
* option of the {@link OutputSpec}.
|
||||
*
|
||||
* <p>Files are streamed to disk (so large outputs are not buffered) and given unique names to avoid
|
||||
* clobbering existing files. The returned {@link ResultFile}s describe what was written (path +
|
||||
* size); they carry a synthetic id because the deliverable is the file on disk, not a {@code
|
||||
* FileStorage} entry, so folder outputs are not downloadable via {@code /files/{id}}.
|
||||
* Writes a run's outputs to the {@code directory} given in the {@link OutputSpec}. Files are
|
||||
* streamed (not buffered) and uniquely named to avoid clobbering. Returned {@link ResultFile}s
|
||||
* carry a synthetic id since the deliverable is the file on disk, not a {@code FileStorage} entry,
|
||||
* so folder outputs are not downloadable via {@code /files/{id}}.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -95,11 +92,7 @@ public class FolderOutputSink implements PolicyOutputSink {
|
||||
return Path.of(directory.toString());
|
||||
}
|
||||
|
||||
/**
|
||||
* The resource's filename reduced to a bare, traversal-free name: any directory component or
|
||||
* "../" is stripped so a crafted output name cannot escape {@code targetDir}. Falls back to a
|
||||
* synthetic name when the filename is absent or reduces to nothing usable.
|
||||
*/
|
||||
// Strip any directory component / "../" so a crafted output name cannot escape targetDir.
|
||||
private static String safeName(String filename, int index) {
|
||||
if (filename == null || filename.isBlank()) {
|
||||
return "output-" + index;
|
||||
@@ -111,7 +104,7 @@ public class FolderOutputSink implements PolicyOutputSink {
|
||||
return name;
|
||||
}
|
||||
|
||||
/** Resolve a non-colliding path in {@code dir}, appending " (n)" before the extension. */
|
||||
// Non-colliding path, appending " (n)" before the extension.
|
||||
private static Path uniqueTarget(Path dir, String filename) {
|
||||
Path candidate = dir.resolve(filename);
|
||||
if (!Files.exists(candidate)) {
|
||||
|
||||
+2
-3
@@ -17,9 +17,8 @@ import stirling.software.common.service.FileStorage;
|
||||
import stirling.software.proprietary.policy.model.OutputSpec;
|
||||
|
||||
/**
|
||||
* Default output sink: stores each output file in {@code FileStorage} so it is downloadable via
|
||||
* {@code GET /api/v1/general/files/{fileId}}. This is the destination for manually-triggered runs
|
||||
* whose results are returned to the caller.
|
||||
* Default sink: stores each output in {@code FileStorage} so it is downloadable via {@code GET
|
||||
* /api/v1/general/files/{fileId}}. Used for manual runs whose results return to the caller.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
|
||||
+5
-16
@@ -9,11 +9,9 @@ import stirling.software.common.model.job.ResultFile;
|
||||
import stirling.software.proprietary.policy.model.OutputSpec;
|
||||
|
||||
/**
|
||||
* Delivers a finished run's output files to a destination, returning durable {@link ResultFile}
|
||||
* descriptors (fileId + metadata) for the run record.
|
||||
*
|
||||
* <p>Implementations are Spring beans selected by {@link #supports(OutputSpec)}. New destinations
|
||||
* (folder, S3) are added as new beans without changing the engine.
|
||||
* Delivers a finished run's outputs to a destination, returning {@link ResultFile} descriptors for
|
||||
* the run record. Implementations are beans selected by {@link #supports(OutputSpec)}, so a new
|
||||
* destination (folder, S3) is just a new bean.
|
||||
*/
|
||||
public interface PolicyOutputSink {
|
||||
|
||||
@@ -23,19 +21,10 @@ public interface PolicyOutputSink {
|
||||
/** Whether this sink can handle the given output spec. */
|
||||
boolean supports(OutputSpec spec);
|
||||
|
||||
/**
|
||||
* Check that an output spec is usable, throwing {@link IllegalArgumentException} if not. Called
|
||||
* when a policy is saved so misconfiguration fails fast rather than at run time.
|
||||
*/
|
||||
/** Throws {@link IllegalArgumentException} on bad config. Called on save to fail fast. */
|
||||
default void validate(OutputSpec spec) {}
|
||||
|
||||
/**
|
||||
* Persist/deliver the output files and return their descriptors.
|
||||
*
|
||||
* @param runId the run these outputs belong to
|
||||
* @param outputs the final pipeline output resources
|
||||
* @param spec the requested destination
|
||||
*/
|
||||
/** Persist/deliver the output files and return their descriptors. */
|
||||
List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
|
||||
throws IOException;
|
||||
}
|
||||
|
||||
+3
-8
@@ -1,22 +1,17 @@
|
||||
package stirling.software.proprietary.policy.progress;
|
||||
|
||||
/**
|
||||
* Receives live progress as a pipeline run executes. Implementations forward to an SSE stream,
|
||||
* write job notes for polling, or both. Step indices are 1-based.
|
||||
*
|
||||
* <p>All methods default to no-ops so callers implement only what they surface.
|
||||
* Receives live progress as a pipeline run executes (SSE stream, job notes, or both). Step indices
|
||||
* are 1-based. All methods default to no-ops.
|
||||
*/
|
||||
public interface PolicyProgressListener {
|
||||
|
||||
/** A listener that ignores all progress. */
|
||||
PolicyProgressListener NOOP = new PolicyProgressListener() {};
|
||||
|
||||
/** Called immediately before step {@code stepIndex} of {@code stepCount} begins. */
|
||||
default void onStepStart(int stepIndex, int stepCount, String operation) {}
|
||||
|
||||
/** Called immediately after step {@code stepIndex} of {@code stepCount} completes. */
|
||||
default void onStepComplete(int stepIndex, int stepCount, String operation) {}
|
||||
|
||||
/** Called on a keep-alive tick so downstream connections can detect disconnects promptly. */
|
||||
/** Keep-alive tick so downstream connections can detect disconnects promptly. */
|
||||
default void onHeartbeat() {}
|
||||
}
|
||||
|
||||
+2
-3
@@ -9,9 +9,8 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* In-memory {@link PolicyStore}. Not the runtime bean - {@link JpaPolicyStore} is the durable
|
||||
* store. Kept as a lightweight, dependency-free implementation for tests and for any future no-
|
||||
* database mode.
|
||||
* In-memory {@link PolicyStore} for tests and any future no-database mode. {@link JpaPolicyStore}
|
||||
* is the runtime bean.
|
||||
*/
|
||||
public class InProcessPolicyStore implements PolicyStore {
|
||||
|
||||
|
||||
+2
-3
@@ -13,9 +13,8 @@ import stirling.software.proprietary.policy.model.Policy;
|
||||
import tools.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Durable {@link PolicyStore} backed by JPA. The runtime store whenever the proprietary module runs
|
||||
* (a datasource is always present). Policies are persisted as JSON via {@link PolicyEntity}; the
|
||||
* scalar columns are kept in sync for querying.
|
||||
* Durable {@link PolicyStore} backed by JPA; the runtime store. Policies are persisted as JSON via
|
||||
* {@link PolicyEntity}, with scalar columns kept in sync for querying.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
|
||||
+5
-7
@@ -12,13 +12,11 @@ import lombok.NoArgsConstructor;
|
||||
import lombok.Setter;
|
||||
|
||||
/**
|
||||
* JPA row for a {@link stirling.software.proprietary.policy.model.Policy}.
|
||||
*
|
||||
* <p>The whole policy is stored as JSON in {@code policyJson} (authoritative on read, and the same
|
||||
* serialization the API uses); the scalar columns are denormalized copies for querying - notably
|
||||
* {@code triggerType} + {@code enabled} so background triggers can fetch their policies. Ownership
|
||||
* is a plain {@code owner} string rather than a foreign key, to stay decoupled from the security
|
||||
* entities; richer team scoping can be layered on later.
|
||||
* JPA row for a {@link stirling.software.proprietary.policy.model.Policy}. The whole policy lives
|
||||
* as JSON in {@code policyJson} (authoritative on read); the scalar columns are denormalized copies
|
||||
* for querying, notably {@code triggerType} + {@code enabled} so background triggers can fetch
|
||||
* their policies. {@code owner} is a plain string, not a foreign key, to stay decoupled from the
|
||||
* security entities.
|
||||
*/
|
||||
@Entity
|
||||
@Table(name = "policies")
|
||||
|
||||
+4
-9
@@ -5,24 +5,19 @@ import java.util.Optional;
|
||||
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* Stores {@link Policy} definitions. The in-memory implementation backs simple deployments now; a
|
||||
* durable (JPA) implementation can replace it behind this interface without touching callers.
|
||||
*/
|
||||
/** Stores {@link Policy} definitions. */
|
||||
public interface PolicyStore {
|
||||
|
||||
/** Create or update a policy. A blank/absent id is assigned; returns the stored policy. */
|
||||
/** Create or update; a blank/absent id is assigned. Returns the stored policy. */
|
||||
Policy save(Policy policy);
|
||||
|
||||
Optional<Policy> get(String id);
|
||||
|
||||
List<Policy> all();
|
||||
|
||||
/**
|
||||
* Enabled policies whose automatic trigger is of the given type (used by background triggers).
|
||||
*/
|
||||
/** Enabled policies with the given trigger type, for background triggers. */
|
||||
List<Policy> findByTriggerType(String triggerType);
|
||||
|
||||
/** Remove a policy; returns whether it existed. */
|
||||
/** Returns whether the policy existed. */
|
||||
boolean delete(String id);
|
||||
}
|
||||
|
||||
+14
-28
@@ -33,20 +33,14 @@ import stirling.software.proprietary.policy.model.Policy;
|
||||
import stirling.software.proprietary.policy.store.PolicyStore;
|
||||
|
||||
/**
|
||||
* Fires policies the moment a file lands in one of their folder sources, instead of polling on a
|
||||
* timer. The trigger only reads that location; turning it into files is still the source's job.
|
||||
* Fires policies when a file lands in one of their folder sources, rather than polling on a timer.
|
||||
*
|
||||
* <p>The watcher is a latency optimisation, not a source of truth, so this pairs an event watch
|
||||
* with a low-frequency <b>reconcile</b> sweep ({@code watchReconcileSeconds}). The reconcile both
|
||||
* (a) re-syncs which directories are watched as policies are created/edited/deleted and folders
|
||||
* appear on disk, and (b) runs every folder-watch policy once, catching files that pre-dated the
|
||||
* watch, events lost to inotify-queue overflow, and changes on filesystems that do not deliver
|
||||
* events at all (NFS, many container bind mounts). Both paths just call {@link PolicyRunner#run};
|
||||
* the {@link InputSource} does the claiming, so a redundant run finds nothing to claim and is
|
||||
* harmless.
|
||||
* <p>The watch is a latency optimisation, not a source of truth: a periodic reconcile sweep ({@code
|
||||
* watchReconcileSeconds}) re-syncs watched dirs and re-runs every policy, covering files that
|
||||
* pre-dated the watch, dropped events, and filesystems that emit none (NFS, bind mounts). Redundant
|
||||
* runs are harmless since {@link InputSource} does the claiming.
|
||||
*
|
||||
* <p>Like {@link ScheduleTrigger}, watch state is per-node and in memory; this assumes a single
|
||||
* node and rebuilds its registrations on restart from the {@link PolicyStore}.
|
||||
* <p>Watch state is in memory, so this assumes a single node and rebuilds registrations on restart.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -65,7 +59,7 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
|
||||
private volatile boolean running;
|
||||
|
||||
// Package-visible (not private) so tests can drive syncRegistrations() against a real service.
|
||||
// Package-visible so tests can drive syncRegistrations() against a real service.
|
||||
volatile WatchService watchService;
|
||||
|
||||
private volatile ScheduledExecutorService reconciler;
|
||||
@@ -125,8 +119,7 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
}
|
||||
|
||||
private void watchLoop() {
|
||||
// Capture the service once: stop() may null the field, and a local avoids racing that to an
|
||||
// NPE (close() still wakes take()/poll() on this same instance).
|
||||
// Capture once: stop() may null the field; close() still wakes take()/poll() on this local.
|
||||
WatchService watcher = watchService;
|
||||
if (watcher == null) {
|
||||
return;
|
||||
@@ -146,9 +139,8 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect the directories touched by {@code first} and any further events that arrive within
|
||||
* the quiet period, so a burst of file-system events becomes a single set of affected
|
||||
* directories. The event kinds are irrelevant: any event on a watched dir just means "go look".
|
||||
* Coalesce a burst of file-system events into one set of affected directories: drain everything
|
||||
* arriving within the quiet period. Event kinds are irrelevant; any event means "go look".
|
||||
*/
|
||||
private Set<Path> drainBurst(WatchService watcher, WatchKey first) {
|
||||
long quietPeriodMs = applicationProperties.getPolicies().getWatchQuietPeriodMs();
|
||||
@@ -200,7 +192,7 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
}
|
||||
}
|
||||
|
||||
/** The reconcile safety net: run every folder-watch policy regardless of watch events. */
|
||||
/** Reconcile safety net: run every folder-watch policy regardless of watch events. */
|
||||
void runAll() {
|
||||
for (Policy policy : policyStore.findByTriggerType(TYPE)) {
|
||||
try {
|
||||
@@ -214,10 +206,7 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Bring the set of watched directories in line with the current folder-watch policies: register
|
||||
* newly required directories that exist on disk, and cancel ones no longer wanted.
|
||||
*/
|
||||
/** Register newly-wanted dirs that exist on disk, cancel ones no longer wanted. */
|
||||
synchronized void syncRegistrations() {
|
||||
if (watchService == null) {
|
||||
return;
|
||||
@@ -274,11 +263,8 @@ public class FolderWatchTrigger implements PolicyTrigger {
|
||||
return dirs;
|
||||
}
|
||||
|
||||
/**
|
||||
* The normalised, absolute directories this policy's sources expose to watch. Normalisation
|
||||
* makes registration keys and event-time matching comparable regardless of how the path was
|
||||
* configured.
|
||||
*/
|
||||
// Absolute + normalised so registration keys and event-time matching compare regardless of how
|
||||
// the path was configured.
|
||||
private List<Path> watchDirsOf(Policy policy) {
|
||||
List<Path> dirs = new ArrayList<>();
|
||||
for (InputSpec spec : policy.sources()) {
|
||||
|
||||
+5
-20
@@ -3,36 +3,21 @@ package stirling.software.proprietary.policy.trigger;
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* An automatic trigger: the thing that decides <em>when</em> a policy runs without a person asking.
|
||||
* A trigger owns a {@link #type()} (matching {@code TriggerConfig.type()}); when its condition
|
||||
* fires it hands the policy to the {@code PolicyRunner}, which pulls the policy's sources and
|
||||
* starts the runs. A trigger never resolves sources itself.
|
||||
*
|
||||
* <p>Triggers are background, configuration-driven beans (schedule, and in future webhook or
|
||||
* folder-watch): on {@link #start()} they begin watching/scheduling for the policies returned by
|
||||
* {@code PolicyStore.findByTriggerType(type())}, and stop on {@link #stop()}. New trigger kinds are
|
||||
* new beans of this type; the runner and the {@code Policy} model do not change.
|
||||
*
|
||||
* <p>Manual running is not a trigger - every policy can always be run on demand via the {@code
|
||||
* PolicyRunner} regardless of whether it has a trigger.
|
||||
* Decides <em>when</em> a policy runs. On firing it hands the policy to {@code PolicyRunner}; it
|
||||
* never resolves sources itself. New trigger kinds are just new beans of this type.
|
||||
*/
|
||||
public interface PolicyTrigger {
|
||||
|
||||
/** Stable identifier for this trigger kind, matching {@code TriggerConfig.type()}. */
|
||||
/** Matches {@code TriggerConfig.type()}. */
|
||||
String type();
|
||||
|
||||
/**
|
||||
* Check that this trigger is usable for the given policy, throwing {@link
|
||||
* IllegalArgumentException} if not. Called when a policy is saved so misconfiguration fails
|
||||
* fast rather than at fire time. Receives the whole {@link Policy} (not just its {@code
|
||||
* TriggerConfig}) so a trigger whose firing depends on the policy's sources (folder-watch) can
|
||||
* assert that relationship; most triggers only inspect {@code policy.trigger()}.
|
||||
* Validate at save time so misconfiguration fails fast, not at fire time. Receives the whole
|
||||
* {@link Policy} so triggers that depend on the policy's sources (folder-watch) can check that.
|
||||
*/
|
||||
default void validate(Policy policy) {}
|
||||
|
||||
/** Begin activating policies of this type (e.g. start the schedule sweep). */
|
||||
default void start() {}
|
||||
|
||||
/** Stop activating and release any resources. */
|
||||
default void stop() {}
|
||||
}
|
||||
|
||||
+1
-8
@@ -8,14 +8,7 @@ import org.springframework.stereotype.Service;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* Starts and stops every {@link PolicyTrigger} with the application lifecycle. Background triggers
|
||||
* (schedule, and future folder/S3) begin watching on {@link #start()} and release resources on
|
||||
* {@link #stop()}; request-driven triggers (manual) are no-ops.
|
||||
*
|
||||
* <p>This is the single activation point for triggers - a new background trigger only has to be a
|
||||
* {@link PolicyTrigger} bean.
|
||||
*/
|
||||
/** Starts and stops every {@link PolicyTrigger} with the application lifecycle. */
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
|
||||
+5
-14
@@ -24,16 +24,9 @@ import stirling.software.proprietary.policy.store.PolicyStore;
|
||||
import tools.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Fires policies on a {@link Schedule}. On {@link #start()} it sweeps on a fixed interval; each
|
||||
* sweep finds the enabled "schedule" policies and runs any whose next firing has come due since it
|
||||
* last fired.
|
||||
* Fires policies on a {@link Schedule}: a fixed-interval sweep runs each due "schedule" policy.
|
||||
*
|
||||
* <p>The trigger only decides <em>when</em>: once a policy is due it hands it to the {@link
|
||||
* PolicyRunner}, which pulls from the policy's configured sources and starts the runs. The trigger
|
||||
* knows nothing about folders, buckets, or how many runs a sweep produces.
|
||||
*
|
||||
* <p>Caveat: last-fire times are tracked <b>in memory</b>, so this assumes a single node and resets
|
||||
* on restart; cluster-wide coordination (leader election) is a follow-up.
|
||||
* <p>Last-fire times are in memory, so this assumes a single node and resets on restart.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@@ -101,8 +94,7 @@ public class ScheduleTrigger implements PolicyTrigger {
|
||||
continue;
|
||||
}
|
||||
|
||||
// First time we see a policy, baseline its last-fire to now so it does not fire
|
||||
// immediately; subsequent sweeps fire it once its next firing has passed.
|
||||
// Baseline a newly-seen policy to now so it does not fire immediately.
|
||||
Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now);
|
||||
ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone()));
|
||||
if (!next.toInstant().isAfter(now)) {
|
||||
@@ -114,9 +106,8 @@ public class ScheduleTrigger implements PolicyTrigger {
|
||||
}
|
||||
|
||||
/**
|
||||
* The typed, validated form of a schedule trigger's options: the {@link Schedule} and the zone
|
||||
* its wall-clock kinds are evaluated in (default UTC). Construction fails for a missing/invalid
|
||||
* schedule or zone.
|
||||
* Validated schedule-trigger options: the {@link Schedule} and the zone it runs in (UTC by
|
||||
* default).
|
||||
*/
|
||||
record ScheduleConfig(Schedule schedule, ZoneId zone) {
|
||||
|
||||
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
package stirling.software.proprietary.policy.config;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
|
||||
import stirling.software.common.model.ApplicationProperties;
|
||||
import stirling.software.common.service.UserServiceInterface;
|
||||
import stirling.software.proprietary.policy.model.OutputSpec;
|
||||
import stirling.software.proprietary.policy.model.Policy;
|
||||
|
||||
/**
|
||||
* Tests for {@link PolicyAccessGuard}: owner-or-admin access, no-op when login is disabled, and
|
||||
* server-side owner assignment.
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class PolicyAccessGuardTest {
|
||||
|
||||
@Mock private UserServiceInterface userService;
|
||||
|
||||
private PolicyAccessGuard guard(boolean loginEnabled) {
|
||||
ApplicationProperties properties = new ApplicationProperties();
|
||||
properties.getSecurity().setEnableLogin(loginEnabled);
|
||||
return new PolicyAccessGuard(userService, properties);
|
||||
}
|
||||
|
||||
@Test
|
||||
void loginDisabledAllowsEverythingAndAssignsNoOwner() {
|
||||
PolicyAccessGuard guard = guard(false);
|
||||
List<Policy> all = List.of(ownedBy("alice"), ownedBy("bob"));
|
||||
|
||||
assertTrue(guard.canAccess(ownedBy("someone-else")));
|
||||
assertNull(guard.ownerForNewPolicy());
|
||||
assertEquals(all, guard.visible(all));
|
||||
}
|
||||
|
||||
@Test
|
||||
void adminCanAccessAnyPolicy() {
|
||||
when(userService.isCurrentUserAdmin()).thenReturn(true);
|
||||
assertTrue(guard(true).canAccess(ownedBy("alice")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void ownerCanAccessTheirOwnPolicy() {
|
||||
when(userService.isCurrentUserAdmin()).thenReturn(false);
|
||||
when(userService.getCurrentUsername()).thenReturn("alice");
|
||||
assertTrue(guard(true).canAccess(ownedBy("alice")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void nonOwnerNonAdminCannotAccess() {
|
||||
when(userService.isCurrentUserAdmin()).thenReturn(false);
|
||||
when(userService.getCurrentUsername()).thenReturn("bob");
|
||||
assertFalse(guard(true).canAccess(ownedBy("alice")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void visibleFiltersToOwnedPoliciesForANonAdmin() {
|
||||
when(userService.isCurrentUserAdmin()).thenReturn(false);
|
||||
when(userService.getCurrentUsername()).thenReturn("alice");
|
||||
|
||||
List<Policy> visible =
|
||||
guard(true).visible(List.of(ownedBy("alice"), ownedBy("bob"), ownedBy("alice")));
|
||||
|
||||
assertEquals(2, visible.size());
|
||||
assertTrue(visible.stream().allMatch(policy -> "alice".equals(policy.owner())));
|
||||
}
|
||||
|
||||
@Test
|
||||
void ownerForNewPolicyIsTheCurrentUserWhenEnforced() {
|
||||
when(userService.getCurrentUsername()).thenReturn("alice");
|
||||
assertEquals("alice", guard(true).ownerForNewPolicy());
|
||||
}
|
||||
|
||||
private static Policy ownedBy(String owner) {
|
||||
return new Policy("p1", "p", owner, true, null, List.of(), List.of(), OutputSpec.inline());
|
||||
}
|
||||
}
|
||||
@@ -3077,14 +3077,6 @@ addTo = "Add to"
|
||||
hint = "Pick your client, paste the snippet into the file shown, then restart it. You'll sign in with your Stirling account on first use - no keys to copy."
|
||||
title = "Connect your AI assistant"
|
||||
|
||||
[config.mcp.tools]
|
||||
ai = "AI"
|
||||
convert = "Convert"
|
||||
misc = "Misc"
|
||||
pages = "Pages"
|
||||
security = "Security"
|
||||
title = "What your assistant can do"
|
||||
|
||||
[config.overview]
|
||||
description = "Current application settings and configuration details."
|
||||
error = "Error"
|
||||
@@ -8193,9 +8185,6 @@ confirm = "Extract"
|
||||
message = "This ZIP contains {{count}} files. Extract anyway?"
|
||||
title = "Large ZIP File"
|
||||
|
||||
[policies]
|
||||
sidebarTitle = "Policies"
|
||||
|
||||
[watchedFolders]
|
||||
alreadyInFolder = "Already in this folder"
|
||||
deleteConfirmBody = "This will remove the folder and its run history. Files already downloaded are not affected."
|
||||
|
||||
@@ -74,7 +74,11 @@ export async function runPolicyPipeline(
|
||||
): Promise<string> {
|
||||
const form = new FormData();
|
||||
for (const file of files) form.append("fileInput", file);
|
||||
form.append("json", JSON.stringify(definition));
|
||||
// The backend binds this as a typed @RequestPart, so it must be an application/json part.
|
||||
form.append(
|
||||
"json",
|
||||
new Blob([JSON.stringify(definition)], { type: "application/json" }),
|
||||
);
|
||||
const res = await apiClient.post<JobResponse>("/api/v1/policies/run", form, {
|
||||
headers: { "Content-Type": "multipart/form-data" },
|
||||
});
|
||||
|
||||
@@ -133,9 +133,10 @@ describe("buildBackendPolicy", () => {
|
||||
expect(policy.steps).toEqual([
|
||||
{ operation: "/api/v1/misc/compress-pdf", parameters: {} },
|
||||
]);
|
||||
// Extras ride in options.
|
||||
expect(policy.trigger.options.categoryId).toBe("security");
|
||||
expect(policy.trigger.options.reviewerEmail).toBe("[email protected]");
|
||||
// Trigger is null (browser-driven); extras ride in output.options.
|
||||
expect(policy.trigger).toBeNull();
|
||||
expect(policy.output.options.categoryId).toBe("security");
|
||||
expect(policy.output.options.reviewerEmail).toBe("[email protected]");
|
||||
expect(policy.output.options.maxRetries).toBe(2);
|
||||
});
|
||||
|
||||
|
||||
@@ -35,7 +35,7 @@ export interface BackendPipelineDefinition {
|
||||
output: BackendOutputSpec;
|
||||
}
|
||||
|
||||
/** How a stored policy is triggered ("manual" | "folder" | "schedule" | "s3"). */
|
||||
/** A policy's automatic trigger ("schedule" | "folder-watch"); null means manual (run on demand). */
|
||||
export interface BackendTriggerConfig {
|
||||
type: string;
|
||||
options: Record<string, unknown>;
|
||||
@@ -49,7 +49,8 @@ export interface BackendPolicy {
|
||||
owner: string;
|
||||
/** Gates automatic triggering; an explicit run ignores it. */
|
||||
enabled: boolean;
|
||||
trigger: BackendTriggerConfig;
|
||||
/** Null = manual; these policies are run from the browser on uploaded files. */
|
||||
trigger: BackendTriggerConfig | null;
|
||||
steps: BackendPipelineStep[];
|
||||
output: BackendOutputSpec;
|
||||
}
|
||||
@@ -188,7 +189,7 @@ export interface PolicyToStore {
|
||||
/** The decoded policy read back from the backend. */
|
||||
export interface DecodedPolicy {
|
||||
id: string;
|
||||
/** The catalog category this policy maps to (from trigger.options.categoryId). */
|
||||
/** The catalog category this policy maps to (from output.options.categoryId). */
|
||||
categoryId: string;
|
||||
name: string;
|
||||
enabled: boolean;
|
||||
@@ -210,13 +211,11 @@ const DEFAULT_FOLDER: PolicyFolderSettings = {
|
||||
};
|
||||
|
||||
/**
|
||||
* Map a frontend policy to the backend {@link BackendPolicy} for persistence.
|
||||
* The backend models only name/enabled/trigger/steps/output, so the policy-level
|
||||
* extras (categoryId, sources, scope, reviewer, fields) ride in `trigger.options`
|
||||
* and the output + retry settings in `output.options`; the full frontend
|
||||
* automation is stashed in `output.options.automation` for a lossless UI
|
||||
* round-trip (while `steps` carries the endpoint-mapped pipeline the engine
|
||||
* runs, pre-built by the caller).
|
||||
* Map a frontend policy to the backend {@link BackendPolicy} for persistence. Trigger is null:
|
||||
* these policies are run from the browser on uploaded files, not fired by a backend trigger. The
|
||||
* backend models only name/enabled/trigger/steps/output, so all policy-level extras (categoryId,
|
||||
* sources, scope, reviewer, fields, output/retry settings, and the full automation for a lossless
|
||||
* UI round-trip) ride in `output.options`; `steps` carries the endpoint-mapped pipeline.
|
||||
*/
|
||||
export function buildBackendPolicy(input: PolicyToStore): BackendPolicy {
|
||||
return {
|
||||
@@ -224,16 +223,7 @@ export function buildBackendPolicy(input: PolicyToStore): BackendPolicy {
|
||||
name: input.name,
|
||||
owner: "",
|
||||
enabled: input.enabled,
|
||||
trigger: {
|
||||
type: "folder",
|
||||
options: {
|
||||
categoryId: input.categoryId,
|
||||
sources: input.sources,
|
||||
scopeTypes: input.scopeTypes,
|
||||
reviewerEmail: input.reviewerEmail,
|
||||
fieldValues: input.fieldValues,
|
||||
},
|
||||
},
|
||||
trigger: null,
|
||||
steps: input.pipelineSteps,
|
||||
output: {
|
||||
type: "inline",
|
||||
@@ -244,6 +234,11 @@ export function buildBackendPolicy(input: PolicyToStore): BackendPolicy {
|
||||
maxRetries: input.folder.maxRetries,
|
||||
retryDelayMinutes: input.folder.retryDelayMinutes,
|
||||
automation: input.automation,
|
||||
categoryId: input.categoryId,
|
||||
sources: input.sources,
|
||||
scopeTypes: input.scopeTypes,
|
||||
reviewerEmail: input.reviewerEmail,
|
||||
fieldValues: input.fieldValues,
|
||||
},
|
||||
},
|
||||
};
|
||||
@@ -251,7 +246,6 @@ export function buildBackendPolicy(input: PolicyToStore): BackendPolicy {
|
||||
|
||||
/** Decode a stored backend policy back into the frontend settings. */
|
||||
export function fromBackendPolicy(policy: BackendPolicy): DecodedPolicy {
|
||||
const trigger = policy.trigger.options;
|
||||
const output = policy.output.options;
|
||||
const str = (v: unknown, fallback = "") =>
|
||||
typeof v === "string" ? v : fallback;
|
||||
@@ -259,19 +253,17 @@ export function fromBackendPolicy(policy: BackendPolicy): DecodedPolicy {
|
||||
typeof v === "number" ? v : fallback;
|
||||
return {
|
||||
id: policy.id,
|
||||
categoryId: str(trigger.categoryId),
|
||||
categoryId: str(output.categoryId),
|
||||
name: policy.name,
|
||||
enabled: policy.enabled,
|
||||
automation: (output.automation as AutomationConfig | undefined) ?? null,
|
||||
sources: Array.isArray(trigger.sources)
|
||||
? (trigger.sources as string[])
|
||||
sources: Array.isArray(output.sources) ? (output.sources as string[]) : [],
|
||||
scopeTypes: Array.isArray(output.scopeTypes)
|
||||
? (output.scopeTypes as string[])
|
||||
: [],
|
||||
scopeTypes: Array.isArray(trigger.scopeTypes)
|
||||
? (trigger.scopeTypes as string[])
|
||||
: [],
|
||||
reviewerEmail: str(trigger.reviewerEmail),
|
||||
reviewerEmail: str(output.reviewerEmail),
|
||||
fieldValues:
|
||||
(trigger.fieldValues as DecodedPolicy["fieldValues"] | undefined) ?? {},
|
||||
(output.fieldValues as DecodedPolicy["fieldValues"] | undefined) ?? {},
|
||||
folder: {
|
||||
outputMode: output.mode === "new_version" ? "new_version" : "new_file",
|
||||
outputName: str(output.name),
|
||||
|
||||
Reference in New Issue
Block a user