diff --git a/javaTest/src/main/java/com/ft546/wcs/demo/DemoApplication.java b/javaTest/src/main/java/com/ft546/wcs/demo/DemoApplication.java index 5672956..8f5fe6c 100644 --- a/javaTest/src/main/java/com/ft546/wcs/demo/DemoApplication.java +++ b/javaTest/src/main/java/com/ft546/wcs/demo/DemoApplication.java @@ -23,6 +23,7 @@ import com.ft546.wcs.shadow.replay.DecisionEvent; import com.ft546.wcs.shadow.resource.ResourceKey; import com.ft546.wcs.shadow.resource.ResourceLedger; import com.ft546.wcs.shadow.snapshot.TrafficSnapshot; +import com.ft546.wcs.shadow.snapshot.TrafficVehicleSnapshot; import java.io.PrintStream; import java.time.Instant; @@ -161,7 +162,7 @@ public final class DemoApplication { new TrafficMapCompiler.MapDefinition( "FT546", 2L, "ZONE-DEMO", true, "demo-sha", "demo-sha", nodes, edges, new LinkedHashMap(), - Collections.singleton("A"), Collections.singleton("C") + Collections.singleton("B"), Collections.singleton("C") ) ); ResourceLedger ledger = new ResourceLedger(map.getCapacities()); @@ -170,7 +171,14 @@ public final class DemoApplication { map, ledger, new RollingDispatchPolicy() ).run( new TrafficSnapshot(1L, Instant.parse("2026-07-30T05:00:00Z"), - "FT546", 2L, "POLICY-DEMO", Collections.emptyList()), + "FT546", 2L, "POLICY-DEMO", Arrays.asList( + new TrafficVehicleSnapshot("V1", "BOOT-V1", 1L, 1L, + Instant.parse("2026-07-30T04:59:59Z"), "A", "B", + Arrays.asList("A", "B"), true, false, false), + new TrafficVehicleSnapshot("V2", "BOOT-V2", 1L, 1L, + Instant.parse("2026-07-30T04:59:59Z"), "C", null, + Collections.singletonList("C"), true, false, false) + )), Arrays.asList( ShadowCandidate.waiting("SHADOW-1", "V1", false, 2, 5, 1, Collections.singleton(ResourceKey.node("B")), null), diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/application/ShadowTrafficCoordinator.java b/javaTest/src/main/java/com/ft546/wcs/shadow/application/ShadowTrafficCoordinator.java index 90c4344..942ef5b 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/application/ShadowTrafficCoordinator.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/application/ShadowTrafficCoordinator.java @@ -5,19 +5,35 @@ import com.ft546.wcs.shadow.decision.ShadowCandidate; import com.ft546.wcs.shadow.decision.ShadowDecision; import com.ft546.wcs.shadow.map.EffectiveTrafficMap; import com.ft546.wcs.shadow.replay.DecisionEvent; +import com.ft546.wcs.shadow.replay.DecisionReplayInput; import com.ft546.wcs.shadow.resource.ResourceKey; import com.ft546.wcs.shadow.resource.ResourceLedger; import com.ft546.wcs.shadow.snapshot.TrafficSnapshot; +import com.ft546.wcs.shadow.snapshot.TrafficVehicleSnapshot; +import com.ft546.wcs.shadow.snapshot.SnapshotFreshnessPolicy; +import com.ft546.wcs.shadow.snapshot.ControlMode; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; +import java.util.Set; public final class ShadowTrafficCoordinator { private final EffectiveTrafficMap map; private final ResourceLedger authoritativeLedger; private final RollingDispatchPolicy policy; + private final SnapshotFreshnessPolicy freshnessPolicy = + new SnapshotFreshnessPolicy(Duration.ofSeconds(2)); + private final Map previousSnapshots = + new LinkedHashMap(); public ShadowTrafficCoordinator(EffectiveTrafficMap map, ResourceLedger authoritativeLedger, RollingDispatchPolicy policy) { @@ -27,30 +43,66 @@ public final class ShadowTrafficCoordinator { } public DecisionEvent run(TrafficSnapshot snapshot, List candidates) { + if (snapshot.getControlMode() != ControlMode.SHADOW) { + throw new IllegalStateException("coordinator accepts SHADOW snapshots only"); + } if (!map.getMapCode().equals(snapshot.getMapCode()) || map.getVersionId() != snapshot.getMapVersion()) { throw new IllegalArgumentException("snapshot map version mismatch"); } - String hash = hash(snapshot, candidates); + Map current = new LinkedHashMap(); + for (TrafficVehicleSnapshot vehicle : snapshot.getVehicles()) { + SnapshotFreshnessPolicy.SnapshotHealth health = freshnessPolicy.validate( + previousSnapshots.get(vehicle.getVehicleId()), vehicle, snapshot.getDecisionTime()); + if (!health.isHealthy()) { + throw new IllegalStateException("vehicle " + vehicle.getVehicleId() + + " quarantined: " + health.getReasonCode()); + } + current.put(vehicle.getVehicleId(), vehicle); + } + for (ShadowCandidate candidate : candidates) { + if (!current.containsKey(candidate.getVehicleId())) { + throw new IllegalStateException("candidate has no frozen vehicle snapshot " + + candidate.getVehicleId()); + } + } + String hash = computeInputHash(map, authoritativeLedger, snapshot, candidates); List decisions = policy.decide(candidates, authoritativeLedger, map); + previousSnapshots.putAll(current); return new DecisionEvent("shadow-" + snapshot.getTickSeq() + "-" + hash.substring(0, 12), snapshot.getTickSeq(), snapshot.getMapCode(), snapshot.getMapVersion(), - snapshot.getPolicyVersion(), hash, decisions); + snapshot.getPolicyVersion(), hash, decisions, + new DecisionReplayInput(map, authoritativeLedger, snapshot, candidates)); } - private String hash(TrafficSnapshot snapshot, List candidates) { + public static String computeInputHash(EffectiveTrafficMap map, ResourceLedger ledger, + TrafficSnapshot snapshot, List candidates) { StringBuilder canonical = new StringBuilder() .append(snapshot.getTickSeq()).append('|') .append(snapshot.getMapCode()).append('|') .append(snapshot.getMapVersion()).append('|') .append(snapshot.getPolicyVersion()); - for (ShadowCandidate candidate : candidates) { + List capacityKeys = new ArrayList(map.getCapacities().keySet()); + Collections.sort(capacityKeys); + for (ResourceKey key : capacityKeys) { + canonical.append("|CAP:").append(key.value()).append('=') + .append(map.getCapacities().get(key)); + List holders = new ArrayList(ledger.holdersFor(key)); + Collections.sort(holders); + for (String holder : holders) canonical.append(':').append(holder); + } + List ordered = new ArrayList(candidates); + Collections.sort(ordered, Comparator.comparing(ShadowCandidate::getTaskId) + .thenComparing(ShadowCandidate::getVehicleId)); + for (ShadowCandidate candidate : ordered) { canonical.append('|').append(candidate.getTaskId()).append(':') .append(candidate.getVehicleId()).append(':') .append(candidate.isCommitted()).append(':') .append(candidate.getBusinessPriority()).append(':') .append(candidate.getWaitAge()); - for (ResourceKey resource : candidate.getResources()) { + List resources = new ArrayList(candidate.getResources()); + Collections.sort(resources); + for (ResourceKey resource : resources) { canonical.append(':').append(resource.value()); } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java index eeef186..183cc4d 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java @@ -20,8 +20,13 @@ public final class RollingDispatchPolicy { List decisions = new ArrayList(); for (ShadowCandidate candidate : candidates) { if (candidate.isCommitted()) { - decisions.add(decision(candidate, DecisionType.GRANT, "COMMITTED_CONTINUATION", - Collections.emptySet(), Collections.emptySet(), null)); + ResourceLedger.ProtectionResult committed = + simulation.evaluate(candidate.getVehicleId(), candidate.getResources()); + decisions.add(committed.isGranted() + ? decision(candidate, DecisionType.GRANT, "COMMITTED_CONTINUATION", + Collections.emptySet(), Collections.emptySet(), null) + : decision(candidate, DecisionType.HOLD, "COMMITTED_RESOURCE_MISMATCH", + committed.getConflictingResources(), committed.getBlockingVehicles(), null)); continue; } ResourceLedger.ProtectionResult result = @@ -31,6 +36,11 @@ public final class RollingDispatchPolicy { Collections.emptySet(), Collections.emptySet(), null)); continue; } + if (result.isUnknownResource()) { + decisions.add(decision(candidate, DecisionType.HOLD, "UNKNOWN_RESOURCE", + result.getConflictingResources(), Collections.emptySet(), null)); + continue; + } boolean cycle = waits.wouldCreateCycle(candidate.getVehicleId(), result.getBlockingVehicles()); waits.addWait(candidate.getVehicleId(), result.getBlockingVehicles()); if (cycle) { diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java index c4c22ae..83be58f 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java @@ -5,6 +5,7 @@ import com.ft546.wcs.shadow.resource.ResourceKey; import java.util.Collections; import java.util.LinkedHashSet; import java.util.Set; +import java.util.Objects; public final class ShadowDecision { private final String taskId; @@ -36,4 +37,23 @@ public final class ShadowDecision { public Set getConflictingResources() { return conflictingResources; } public Set getBlockingVehicles() { return blockingVehicles; } public String getYieldSite() { return yieldSite; } + + @Override + public boolean equals(Object other) { + if (!(other instanceof ShadowDecision)) return false; + ShadowDecision that = (ShadowDecision) other; + return Objects.equals(taskId, that.taskId) + && Objects.equals(vehicleId, that.vehicleId) + && type == that.type + && Objects.equals(reasonCode, that.reasonCode) + && Objects.equals(conflictingResources, that.conflictingResources) + && Objects.equals(blockingVehicles, that.blockingVehicles) + && Objects.equals(yieldSite, that.yieldSite); + } + + @Override + public int hashCode() { + return Objects.hash(taskId, vehicleId, type, reasonCode, + conflictingResources, blockingVehicles, yieldSite); + } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/map/TrafficMapCompiler.java b/javaTest/src/main/java/com/ft546/wcs/shadow/map/TrafficMapCompiler.java index 53cf3f8..a68a261 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/map/TrafficMapCompiler.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/map/TrafficMapCompiler.java @@ -19,23 +19,46 @@ public final class TrafficMapCompiler { if (!definition.baseChecksum.equals(definition.configuredChecksum)) { throw new IllegalArgumentException("base map checksum mismatch"); } + if (definition.versionId <= 0 || definition.nodes.isEmpty()) { + throw new IllegalArgumentException("positive map version and nodes are required"); + } + if (!definition.nodes.containsAll(definition.safeWaitSites) + || !definition.nodes.containsAll(definition.safeYieldSites)) { + throw new IllegalArgumentException("safe site references missing node"); + } LinkedHashSet edges = new LinkedHashSet(); LinkedHashMap capacities = new LinkedHashMap(); + LinkedHashMap degree = new LinkedHashMap(); for (String node : definition.nodes) { capacities.put(ResourceKey.node(node), 1); + degree.put(node, 0); } for (EdgeDefinition edge : definition.edges) { if (!definition.nodes.contains(edge.first) || !definition.nodes.contains(edge.second)) { throw new IllegalArgumentException("edge references missing node"); } + if (edge.first.equals(edge.second)) { + throw new IllegalArgumentException("self edge is not allowed"); + } ResourceKey key = ResourceKey.edge(edge.first, edge.second); - edges.add(key); + if (!edges.add(key)) throw new IllegalArgumentException("duplicate edge " + key); capacities.put(key, 1); + degree.put(edge.first, degree.get(edge.first) + 1); + degree.put(edge.second, degree.get(edge.second) + 1); + } + for (String waitSite : definition.safeWaitSites) { + if (degree.get(waitSite) <= 1) { + throw new IllegalArgumentException("safe wait site is a unique exit " + waitSite); + } } for (Map.Entry entry : definition.explicitCapacities.entrySet()) { if (entry.getValue() == null || entry.getValue() <= 0) { throw new IllegalArgumentException("resource capacity must be positive"); } + if (!entry.getKey().value().startsWith("CONFLICT:") + && !capacities.containsKey(entry.getKey())) { + throw new IllegalArgumentException("capacity references unknown resource " + entry.getKey()); + } capacities.put(entry.getKey(), entry.getValue()); } return new EffectiveTrafficMap(definition.mapCode, definition.versionId, definition.zoneCode, @@ -87,6 +110,30 @@ public final class TrafficMapCompiler { this.safeYieldSites = immutable(safeYieldSites); } + public MapDefinition withSafeWaitSites(Set sites) { + return new MapDefinition(mapCode, versionId, zoneCode, confirmed, baseChecksum, + configuredChecksum, nodes, edges, explicitCapacities, sites, safeYieldSites); + } + + public MapDefinition withAdditionalEdge(EdgeDefinition edge) { + Set changed = new LinkedHashSet(edges); + changed.add(edge); + return new MapDefinition(mapCode, versionId, zoneCode, confirmed, baseChecksum, + configuredChecksum, nodes, changed, explicitCapacities, safeWaitSites, safeYieldSites); + } + + public MapDefinition withCapacity(ResourceKey key, int capacity) { + Map changed = new LinkedHashMap(explicitCapacities); + changed.put(key, capacity); + return new MapDefinition(mapCode, versionId, zoneCode, confirmed, baseChecksum, + configuredChecksum, nodes, edges, changed, safeWaitSites, safeYieldSites); + } + + public MapDefinition withVersionId(long changedVersionId) { + return new MapDefinition(mapCode, changedVersionId, zoneCode, confirmed, baseChecksum, + configuredChecksum, nodes, edges, explicitCapacities, safeWaitSites, safeYieldSites); + } + private static Set immutable(Set source) { return Collections.unmodifiableSet(new LinkedHashSet(source)); } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionEvent.java b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionEvent.java index 75b2751..2bb6068 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionEvent.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionEvent.java @@ -14,9 +14,16 @@ public final class DecisionEvent { private final String policyVersion; private final String inputHash; private final List decisions; + private final DecisionReplayInput replayInput; public DecisionEvent(String decisionId, long tickSeq, String mapCode, long mapVersion, String policyVersion, String inputHash, List decisions) { + this(decisionId, tickSeq, mapCode, mapVersion, policyVersion, inputHash, decisions, null); + } + + public DecisionEvent(String decisionId, long tickSeq, String mapCode, long mapVersion, + String policyVersion, String inputHash, List decisions, + DecisionReplayInput replayInput) { this.decisionId = decisionId; this.tickSeq = tickSeq; this.mapCode = mapCode; @@ -24,6 +31,7 @@ public final class DecisionEvent { this.policyVersion = policyVersion; this.inputHash = inputHash; this.decisions = Collections.unmodifiableList(new ArrayList(decisions)); + this.replayInput = replayInput; } public String getDecisionId() { return decisionId; } @@ -33,4 +41,15 @@ public final class DecisionEvent { public String getPolicyVersion() { return policyVersion; } public String getInputHash() { return inputHash; } public List getDecisions() { return decisions; } + public DecisionReplayInput getReplayInput() { return replayInput; } + + public DecisionEvent withInputHash(String changedHash) { + return new DecisionEvent(decisionId, tickSeq, mapCode, mapVersion, policyVersion, + changedHash, decisions, replayInput); + } + + public DecisionEvent withDecisions(List changedDecisions) { + return new DecisionEvent(decisionId, tickSeq, mapCode, mapVersion, policyVersion, + inputHash, changedDecisions, replayInput); + } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayInput.java b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayInput.java new file mode 100644 index 0000000..d83e7b9 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayInput.java @@ -0,0 +1,31 @@ +package com.ft546.wcs.shadow.replay; + +import com.ft546.wcs.shadow.decision.ShadowCandidate; +import com.ft546.wcs.shadow.map.EffectiveTrafficMap; +import com.ft546.wcs.shadow.resource.ResourceLedger; +import com.ft546.wcs.shadow.snapshot.TrafficSnapshot; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class DecisionReplayInput { + private final EffectiveTrafficMap map; + private final ResourceLedger ledger; + private final TrafficSnapshot snapshot; + private final List candidates; + + public DecisionReplayInput(EffectiveTrafficMap map, ResourceLedger ledger, + TrafficSnapshot snapshot, List candidates) { + this.map = map; + this.ledger = ledger.copy(); + this.snapshot = snapshot; + this.candidates = Collections.unmodifiableList(new ArrayList(candidates)); + } + + public EffectiveTrafficMap getMap() { return map; } + public ResourceLedger freshLedger() { return ledger.copy(); } + public TrafficSnapshot getSnapshot() { return snapshot; } + public List getCandidates() { return candidates; } + ResourceLedger storedLedger() { return ledger; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayService.java b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayService.java index 446f650..82695ff 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayService.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/replay/DecisionReplayService.java @@ -4,13 +4,24 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.ft546.wcs.shadow.decision.DecisionType; import com.ft546.wcs.shadow.decision.ShadowDecision; +import com.ft546.wcs.shadow.decision.ShadowCandidate; +import com.ft546.wcs.shadow.decision.RollingDispatchPolicy; +import com.ft546.wcs.shadow.application.ShadowTrafficCoordinator; +import com.ft546.wcs.shadow.map.EffectiveTrafficMap; +import com.ft546.wcs.shadow.map.TrafficMapCompiler; import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.resource.ResourceLedger; +import com.ft546.wcs.shadow.snapshot.TrafficSnapshot; import java.io.IOException; import java.io.InputStream; +import java.time.Instant; import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Set; public final class DecisionReplayService { @@ -19,6 +30,66 @@ public final class DecisionReplayService { public DecisionEvent read(InputStream input) throws IOException { if (input == null) throw new IllegalArgumentException("decision event input is required"); JsonNode root = mapper.readTree(input); + requireText(root, "decisionId"); + requireText(root, "mapCode"); + requireText(root, "policyVersion"); + requireText(root, "inputHash"); + requireArray(root, "nodes"); + requireArray(root, "edges"); + requireArray(root, "candidates"); + requireArray(root, "decisions"); + long tickSeq = requirePositive(root, "tickSeq"); + long mapVersion = requirePositive(root, "mapVersion"); + + Set nodes = strings(root.withArray("nodes")); + Set edges = + new LinkedHashSet(); + for (JsonNode edge : root.withArray("edges")) { + if (!edge.isArray() || edge.size() != 2) { + throw new IllegalArgumentException("edge must contain exactly two sites"); + } + edges.add(new TrafficMapCompiler.EdgeDefinition(edge.get(0).asText(), edge.get(1).asText())); + } + Map capacities = new LinkedHashMap(); + JsonNode capacityNode = root.path("explicitCapacities"); + if (!capacityNode.isObject()) throw new IllegalArgumentException("explicitCapacities is required"); + java.util.Iterator> capacityFields = capacityNode.fields(); + while (capacityFields.hasNext()) { + Map.Entry entry = capacityFields.next(); + capacities.put(ResourceKey.fromValue(entry.getKey()), entry.getValue().asInt()); + } + EffectiveTrafficMap map = new TrafficMapCompiler().compile( + new TrafficMapCompiler.MapDefinition(root.path("mapCode").asText(), mapVersion, + requireText(root, "zoneCode"), true, + requireText(root, "checksum"), requireText(root, "checksum"), + nodes, edges, capacities, strings(root.withArray("safeWaitSites")), + strings(root.withArray("safeYieldSites")))); + ResourceLedger ledger = new ResourceLedger(map.getCapacities()); + JsonNode holdersNode = root.path("holders"); + if (!holdersNode.isObject()) throw new IllegalArgumentException("holders is required"); + java.util.Iterator> holderFields = holdersNode.fields(); + while (holderFields.hasNext()) { + Map.Entry entry = holderFields.next(); + ResourceKey resource = ResourceKey.fromValue(entry.getKey()); + for (JsonNode holder : entry.getValue()) { + if (!ledger.evaluate(holder.asText(), Collections.singleton(resource)).isGranted()) { + throw new IllegalArgumentException("holder snapshot exceeds resource capacity"); + } + } + } + List candidates = new ArrayList(); + for (JsonNode candidate : root.withArray("candidates")) { + Set resources = resourceKeys(candidate.withArray("resources")); + candidates.add(candidate.path("committed").asBoolean() + ? ShadowCandidate.committed(requireText(candidate, "taskId"), + requireText(candidate, "vehicleId"), resources) + : ShadowCandidate.waiting(requireText(candidate, "taskId"), + requireText(candidate, "vehicleId"), candidate.path("emergency").asBoolean(), + candidate.path("businessPriority").asInt(), candidate.path("waitAge").asLong(), + candidate.path("clearanceCost").asInt(), resources, + candidate.path("safeYieldSite").isNull() ? null + : candidate.path("safeYieldSite").asText())); + } List decisions = new ArrayList(); for (JsonNode node : root.withArray("decisions")) { Set conflicts = new LinkedHashSet(); @@ -35,12 +106,63 @@ public final class DecisionReplayService { node.path("reasonCode").asText(), conflicts, blockers, node.path("yieldSite").isNull() ? null : node.path("yieldSite").asText())); } - return new DecisionEvent(root.path("decisionId").asText(), root.path("tickSeq").asLong(), - root.path("mapCode").asText(), root.path("mapVersion").asLong(), - root.path("policyVersion").asText(), root.path("inputHash").asText(), decisions); + TrafficSnapshot snapshot = new TrafficSnapshot(tickSeq, Instant.EPOCH, + root.path("mapCode").asText(), mapVersion, + root.path("policyVersion").asText(), Collections.emptyList()); + DecisionReplayInput replayInput = new DecisionReplayInput(map, ledger, snapshot, candidates); + return new DecisionEvent(root.path("decisionId").asText(), tickSeq, + root.path("mapCode").asText(), mapVersion, root.path("policyVersion").asText(), + root.path("inputHash").asText(), decisions, replayInput); } public List replay(DecisionEvent event) { - return event.getDecisions(); + if (event.getReplayInput() == null) { + throw new IllegalStateException("event has no replay input"); + } + DecisionReplayInput input = event.getReplayInput(); + String actualHash = ShadowTrafficCoordinator.computeInputHash( + input.getMap(), input.storedLedger(), input.getSnapshot(), input.getCandidates()); + if (!actualHash.equals(event.getInputHash())) { + throw new IllegalStateException("input hash mismatch expected=" + actualHash + + " actual=" + event.getInputHash()); + } + List recomputed = new RollingDispatchPolicy().decide( + input.getCandidates(), input.freshLedger(), input.getMap()); + if (!recomputed.equals(event.getDecisions())) { + throw new IllegalStateException("recorded decisions do not match replay"); + } + return recomputed; + } + + private static String requireText(JsonNode node, String field) { + JsonNode value = node.get(field); + if (value == null || !value.isTextual() || value.asText().trim().isEmpty()) { + throw new IllegalArgumentException(field + " is required"); + } + return value.asText(); + } + + private static long requirePositive(JsonNode node, String field) { + JsonNode value = node.get(field); + if (value == null || !value.canConvertToLong() || value.asLong() <= 0) { + throw new IllegalArgumentException(field + " must be positive"); + } + return value.asLong(); + } + + private static void requireArray(JsonNode node, String field) { + if (!node.path(field).isArray()) throw new IllegalArgumentException(field + " must be an array"); + } + + private static Set strings(JsonNode array) { + Set result = new LinkedHashSet(); + for (JsonNode value : array) result.add(value.asText()); + return result; + } + + private static Set resourceKeys(JsonNode array) { + Set result = new LinkedHashSet(); + for (JsonNode value : array) result.add(ResourceKey.fromValue(value.asText())); + return result; } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/resource/ResourceLedger.java b/javaTest/src/main/java/com/ft546/wcs/shadow/resource/ResourceLedger.java index 6da73cd..247db78 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/resource/ResourceLedger.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/resource/ResourceLedger.java @@ -20,6 +20,10 @@ public final class ResourceLedger { LinkedHashSet conflicts = new LinkedHashSet(); LinkedHashSet blockers = new LinkedHashSet(); for (ResourceKey key : requested) { + if (!capacities.containsKey(key)) { + conflicts.add(key); + return new ProtectionResult(false, true, conflicts, blockers); + } int capacity = capacity(key); Set current = holders.containsKey(key) ? holders.get(key) : Collections.emptySet(); int occupiedByOthers = current.contains(vehicleId) ? current.size() - 1 : current.size(); @@ -33,7 +37,7 @@ public final class ResourceLedger { } } if (!conflicts.isEmpty()) { - return new ProtectionResult(false, conflicts, blockers); + return new ProtectionResult(false, false, conflicts, blockers); } for (ResourceKey key : requested) { if (!holders.containsKey(key)) { @@ -41,7 +45,8 @@ public final class ResourceLedger { } holders.get(key).add(vehicleId); } - return new ProtectionResult(true, Collections.emptySet(), Collections.emptySet()); + return new ProtectionResult(true, false, + Collections.emptySet(), Collections.emptySet()); } public void release(String vehicleId, Collection resources) { @@ -71,6 +76,19 @@ public final class ResourceLedger { return copy; } + public Map getCapacities() { + return capacities; + } + + public Map> snapshotHolders() { + Map> snapshot = new LinkedHashMap>(); + for (Map.Entry> entry : holders.entrySet()) { + snapshot.put(entry.getKey(), Collections.unmodifiableSet( + new LinkedHashSet(entry.getValue()))); + } + return Collections.unmodifiableMap(snapshot); + } + private int capacity(ResourceKey key) { Integer capacity = capacities.get(key); if (capacity == null) { @@ -81,12 +99,15 @@ public final class ResourceLedger { public static final class ProtectionResult { private final boolean granted; + private final boolean unknownResource; private final Set conflictingResources; private final Set blockingVehicles; - private ProtectionResult(boolean granted, Set conflictingResources, + private ProtectionResult(boolean granted, boolean unknownResource, + Set conflictingResources, Set blockingVehicles) { this.granted = granted; + this.unknownResource = unknownResource; this.conflictingResources = Collections.unmodifiableSet( new LinkedHashSet(conflictingResources)); this.blockingVehicles = Collections.unmodifiableSet( @@ -94,6 +115,7 @@ public final class ResourceLedger { } public boolean isGranted() { return granted; } + public boolean isUnknownResource() { return unknownResource; } public Set getConflictingResources() { return conflictingResources; } public Set getBlockingVehicles() { return blockingVehicles; } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/ControlMode.java b/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/ControlMode.java new file mode 100644 index 0000000..114466c --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/ControlMode.java @@ -0,0 +1,5 @@ +package com.ft546.wcs.shadow.snapshot; + +public enum ControlMode { + LEGACY, SHADOW, ACTIVE +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/TrafficSnapshot.java b/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/TrafficSnapshot.java index ef5cf89..e887f61 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/TrafficSnapshot.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/snapshot/TrafficSnapshot.java @@ -11,15 +11,23 @@ public final class TrafficSnapshot { private final String mapCode; private final long mapVersion; private final String policyVersion; + private final ControlMode controlMode; private final List vehicles; public TrafficSnapshot(long tickSeq, Instant decisionTime, String mapCode, long mapVersion, String policyVersion, List vehicles) { + this(tickSeq, decisionTime, mapCode, mapVersion, policyVersion, ControlMode.SHADOW, vehicles); + } + + public TrafficSnapshot(long tickSeq, Instant decisionTime, String mapCode, long mapVersion, + String policyVersion, ControlMode controlMode, + List vehicles) { this.tickSeq = tickSeq; this.decisionTime = decisionTime; this.mapCode = mapCode; this.mapVersion = mapVersion; this.policyVersion = policyVersion; + this.controlMode = controlMode; this.vehicles = Collections.unmodifiableList(new ArrayList(vehicles)); } @@ -28,5 +36,6 @@ public final class TrafficSnapshot { public String getMapCode() { return mapCode; } public long getMapVersion() { return mapVersion; } public String getPolicyVersion() { return policyVersion; } + public ControlMode getControlMode() { return controlMode; } public List getVehicles() { return vehicles; } } diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/MapAndResourceLedgerTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/MapAndResourceLedgerTest.java index 92b3203..c3e6e66 100644 --- a/javaTest/src/test/java/com/ft546/wcs/shadow/MapAndResourceLedgerTest.java +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/MapAndResourceLedgerTest.java @@ -40,6 +40,20 @@ class MapAndResourceLedgerTest { () -> compiler.compile(definition("sha-546", "sha-546", false))); } + @Test + void rejectsUnsafeWaitSitesSelfEdgesAndUnknownCapacityResources() { + TrafficMapCompiler compiler = new TrafficMapCompiler(); + TrafficMapCompiler.MapDefinition base = definition("sha-546", "sha-546", true); + + assertThrows(IllegalArgumentException.class, () -> compiler.compile( + base.withSafeWaitSites(Collections.singleton("A")))); + assertThrows(IllegalArgumentException.class, () -> compiler.compile( + base.withAdditionalEdge(new TrafficMapCompiler.EdgeDefinition("B", "B")))); + assertThrows(IllegalArgumentException.class, () -> compiler.compile( + base.withCapacity(ResourceKey.node("Z"), 1))); + assertThrows(IllegalArgumentException.class, () -> compiler.compile(base.withVersionId(0L))); + } + @Test void enforcesNodeEdgeAndConfiguredCapacity() { EffectiveTrafficMap map = new TrafficMapCompiler().compile(definition("sha-546", "sha-546", true)); @@ -72,6 +86,6 @@ class MapAndResourceLedgerTest { return new TrafficMapCompiler.MapDefinition( "FT546", 2L, "ZONE-1", confirmed, baseChecksum, configuredChecksum, nodes, edges, capacities, - Collections.singleton("A"), Collections.singleton("C")); + Collections.singleton("B"), Collections.singleton("C")); } } diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java index e0e7913..315acdb 100644 --- a/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java @@ -66,7 +66,7 @@ class RollingDispatchPolicyTest { ShadowCandidate first = candidate("T1", "V1", 5, 20, ResourceKey.node("A"), null); ShadowCandidate second = candidate("T2", "V2", 5, 10, ResourceKey.node("B"), null); ShadowCandidate committed = ShadowCandidate.committed( - "T0", "V0", Collections.singleton(ResourceKey.node("A"))); + "T0", "V0", Collections.singleton(ResourceKey.node("C"))); List decisions = new RollingDispatchPolicy().decide(Arrays.asList(first, second, committed), ledger, map); @@ -76,6 +76,22 @@ class RollingDispatchPolicyTest { assertEquals("WAIT_CYCLE", decisions.get(2).getReasonCode()); } + @Test + void committedSegmentProtectsResourcesBeforeNewCandidatesCompete() { + EffectiveTrafficMap map = map(); + ShadowCandidate committed = ShadowCandidate.committed( + "T0", "V0", Collections.singleton(ResourceKey.node("A"))); + ShadowCandidate waiting = candidate( + "T1", "V1", 9, 100, ResourceKey.node("A"), null); + + List decisions = new RollingDispatchPolicy().decide( + Arrays.asList(waiting, committed), new ResourceLedger(map.getCapacities()), map); + + assertEquals(DecisionType.GRANT, decisions.get(0).getType()); + assertEquals(DecisionType.HOLD, decisions.get(1).getType()); + assertEquals(Collections.singleton("V0"), decisions.get(1).getBlockingVehicles()); + } + @Test void detectsCycleBeforeAddingWaitEdges() { WaitForGraph graph = new WaitForGraph(); @@ -85,6 +101,20 @@ class RollingDispatchPolicyTest { assertFalse(graph.isAcyclic()); } + @Test + void unknownResourceFailsClosedAsStructuredHold() { + EffectiveTrafficMap map = map(); + ShadowCandidate candidate = candidate( + "T1", "V1", 1, 1, ResourceKey.conflict("UNPUBLISHED"), null); + + ShadowDecision decision = new RollingDispatchPolicy().decide( + Collections.singletonList(candidate), + new ResourceLedger(map.getCapacities()), map).get(0); + + assertEquals(DecisionType.HOLD, decision.getType()); + assertEquals("UNKNOWN_RESOURCE", decision.getReasonCode()); + } + private ShadowCandidate candidate(String task, String vehicle, int priority, long age, ResourceKey resource, String yieldSite) { return ShadowCandidate.waiting(task, vehicle, false, priority, age, 1, @@ -99,7 +129,7 @@ class RollingDispatchPolicyTest { edges.add(new TrafficMapCompiler.EdgeDefinition("B", "C")); return new TrafficMapCompiler().compile(new TrafficMapCompiler.MapDefinition( "FT546", 2L, "ZONE-1", true, "sha", "sha", nodes, edges, - new LinkedHashMap(), Collections.singleton("A"), + new LinkedHashMap(), Collections.singleton("B"), Collections.singleton("C"))); } } diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowCoordinatorReplayTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowCoordinatorReplayTest.java index 5f9611b..ab8b51d 100644 --- a/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowCoordinatorReplayTest.java +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowCoordinatorReplayTest.java @@ -13,6 +13,8 @@ import com.ft546.wcs.shadow.replay.DecisionReplayService; import com.ft546.wcs.shadow.resource.ResourceKey; import com.ft546.wcs.shadow.resource.ResourceLedger; import com.ft546.wcs.shadow.snapshot.TrafficSnapshot; +import com.ft546.wcs.shadow.snapshot.TrafficVehicleSnapshot; +import com.ft546.wcs.shadow.snapshot.ControlMode; import org.junit.jupiter.api.Test; import java.io.InputStream; @@ -45,19 +47,51 @@ class ShadowCoordinatorReplayTest { new ShadowTrafficCoordinator(map, ledger, new RollingDispatchPolicy()); TrafficSnapshot snapshot = new TrafficSnapshot(9L, Instant.parse("2026-07-30T05:00:00Z"), - "FT546", 2L, "POLICY-1", Collections.emptyList()); + "FT546", 2L, "POLICY-1", Collections.singletonList(vehicle( + "V1", "BOOT-1", 1L, Instant.parse("2026-07-30T04:59:59Z")))); ShadowCandidate candidate = ShadowCandidate.waiting( "T1", "V1", false, 5, 12, 1, Collections.singleton(ResourceKey.node("B")), null); DecisionEvent first = coordinator.run(snapshot, Collections.singletonList(candidate)); - DecisionEvent second = coordinator.run(snapshot, Collections.singletonList(candidate)); + DecisionEvent second = new ShadowTrafficCoordinator( + map, ledger, new RollingDispatchPolicy()).run(snapshot, Collections.singletonList(candidate)); assertEquals(first.getInputHash(), second.getInputHash()); assertEquals(DecisionType.HOLD, first.getDecisions().get(0).getType()); assertEquals(1, ledger.protectedCount(ResourceKey.node("B"))); } + @Test + void rejectsNonShadowStaleDisconnectedAndRestartedSnapshots() { + EffectiveTrafficMap map = map(); + ShadowCandidate candidate = ShadowCandidate.waiting( + "T1", "V1", false, 1, 1, 1, + Collections.singleton(ResourceKey.node("A")), null); + ShadowTrafficCoordinator coordinator = new ShadowTrafficCoordinator( + map, new ResourceLedger(map.getCapacities()), new RollingDispatchPolicy()); + Instant now = Instant.parse("2026-07-30T05:00:05Z"); + + assertThrows(IllegalStateException.class, () -> coordinator.run( + new TrafficSnapshot(1L, now, "FT546", 2L, "P", ControlMode.LEGACY, + Collections.singletonList(vehicle("V1", "BOOT-1", 1L, now))), + Collections.singletonList(candidate))); + assertThrows(IllegalStateException.class, () -> coordinator.run( + new TrafficSnapshot(2L, now, "FT546", 2L, "P", + Collections.singletonList(vehicle("V1", "BOOT-1", 2L, now.minusSeconds(3)))), + Collections.singletonList(candidate))); + + ShadowTrafficCoordinator restart = new ShadowTrafficCoordinator( + map, new ResourceLedger(map.getCapacities()), new RollingDispatchPolicy()); + restart.run(new TrafficSnapshot(3L, now, "FT546", 2L, "P", + Collections.singletonList(vehicle("V1", "BOOT-1", 3L, now))), + Collections.singletonList(candidate)); + assertThrows(IllegalStateException.class, () -> restart.run( + new TrafficSnapshot(4L, now.plusSeconds(1), "FT546", 2L, "P", + Collections.singletonList(vehicle("V1", "BOOT-2", 4L, now.plusSeconds(1)))), + Collections.singletonList(candidate))); + } + @Test void readsAndReplaysGoldenDecisionEvent() throws Exception { InputStream input = getClass().getResourceAsStream("/shadow/golden/basic-conflict.json"); @@ -72,6 +106,21 @@ class ShadowCoordinatorReplayTest { replay.replay(event).get(0).getConflictingResources()); } + @Test + void replayRejectsTamperedHashOrRecordedDecision() throws Exception { + DecisionReplayService replay = new DecisionReplayService(); + DecisionEvent event = replay.read( + getClass().getResourceAsStream("/shadow/golden/basic-conflict.json")); + DecisionEvent tamperedHash = event.withInputHash("tampered"); + DecisionEvent tamperedDecision = event.withDecisions(Collections.singletonList( + new com.ft546.wcs.shadow.decision.ShadowDecision( + "T1", "V1", DecisionType.GRANT, "RESOURCES_AVAILABLE", + Collections.emptySet(), Collections.emptySet(), null))); + + assertThrows(IllegalStateException.class, () -> replay.replay(tamperedHash)); + assertThrows(IllegalStateException.class, () -> replay.replay(tamperedDecision)); + } + private EffectiveTrafficMap map() { LinkedHashSet nodes = new LinkedHashSet(Arrays.asList("A", "B", "C")); LinkedHashSet edges = @@ -80,7 +129,12 @@ class ShadowCoordinatorReplayTest { edges.add(new TrafficMapCompiler.EdgeDefinition("B", "C")); return new TrafficMapCompiler().compile(new TrafficMapCompiler.MapDefinition( "FT546", 2L, "ZONE-1", true, "sha", "sha", nodes, edges, - new LinkedHashMap(), Collections.singleton("A"), + new LinkedHashMap(), Collections.singleton("B"), Collections.singleton("C"))); } + + private TrafficVehicleSnapshot vehicle(String id, String boot, long seq, Instant at) { + return new TrafficVehicleSnapshot(id, boot, seq, seq, at, "A", "B", + Arrays.asList("A", "B"), true, false, false); + } } diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowInvariantPropertyTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowInvariantPropertyTest.java index 326c10f..c95581c 100644 --- a/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowInvariantPropertyTest.java +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/ShadowInvariantPropertyTest.java @@ -102,7 +102,7 @@ class ShadowInvariantPropertyTest { edges.add(new TrafficMapCompiler.EdgeDefinition("N2", "N3")); return new TrafficMapCompiler().compile(new TrafficMapCompiler.MapDefinition( "FT546", 2L, "ZONE-PROPERTY", true, "sha", "sha", nodes, edges, - new LinkedHashMap(), Collections.singleton("N0"), + new LinkedHashMap(), Collections.singleton("N1"), Collections.singleton("N3"))); } } diff --git a/javaTest/src/test/resources/shadow/golden/basic-conflict.json b/javaTest/src/test/resources/shadow/golden/basic-conflict.json index a8662e7..36fceda 100644 --- a/javaTest/src/test/resources/shadow/golden/basic-conflict.json +++ b/javaTest/src/test/resources/shadow/golden/basic-conflict.json @@ -3,8 +3,31 @@ "tickSeq": 9, "mapCode": "FT546", "mapVersion": 2, + "zoneCode": "ZONE-1", + "checksum": "sha", "policyVersion": "POLICY-1", - "inputHash": "golden-input-sha256", + "inputHash": "8526c0ec7be7385b0a69662132c0c8bd31501b98d880155e9c125b51f077819a", + "nodes": ["A", "B", "C"], + "edges": [["A", "B"], ["B", "C"]], + "explicitCapacities": {}, + "safeWaitSites": ["B"], + "safeYieldSites": ["C"], + "holders": { + "NODE:B": ["BLOCKER"] + }, + "candidates": [ + { + "taskId": "T1", + "vehicleId": "V1", + "committed": false, + "emergency": false, + "businessPriority": 5, + "waitAge": 12, + "clearanceCost": 1, + "resources": ["NODE:B"], + "safeYieldSite": null + } + ], "decisions": [ { "taskId": "T1",