diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java index b0289a37..f55dafbe 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java @@ -27,11 +27,11 @@ public class GoalAttemptStore { jdbc.update(""" INSERT INTO mate_goal_attempt( attempt_id,goal_id,conversation_id,parent_attempt_id,trigger_type,state, - lease_token,lease_until,input_item_id,replay_safety,checkpoint_type, + lease_token,lease_until,lease_until_epoch_second,input_item_id,replay_safety,checkpoint_type, created_at,updated_at) - VALUES(?,?,?,?,?,'claimed',?,?,?,'safe','claimed',?,?) + VALUES(?,?,?,?,?,'claimed',?,?,?,?,'safe','claimed',?,?) """, id, goalId, conversationId, parentAttemptId, triggerType, leaseToken, - leaseUntil, inputItemId, now, now); + leaseUntil, GoalLeaseTime.epoch(leaseUntil), inputItemId, now, now); return get(id); } @@ -49,6 +49,13 @@ public class GoalAttemptStore { """, (rs, row) -> read(rs), goalId, Math.max(1, Math.min(limit, 100))); } + public boolean hasLiveFence(String id, String token, LocalDateTime now) { + return jdbc.queryForList(""" + SELECT attempt_id FROM mate_goal_attempt WHERE attempt_id=? AND lease_token=? + AND state IN ('claimed','running') AND lease_until_epoch_second>? FOR UPDATE + """, String.class, id, token, GoalLeaseTime.epoch(now)).size() == 1; + } + public boolean markRunning(String id, String leaseToken, LocalDateTime now) { return jdbc.update(""" UPDATE mate_goal_attempt SET state='running',started_at=?,updated_at=? @@ -59,9 +66,9 @@ public class GoalAttemptStore { public boolean renew(String id, String leaseToken, LocalDateTime leaseUntil, LocalDateTime now) { return jdbc.update(""" - UPDATE mate_goal_attempt SET lease_until=?,updated_at=? + UPDATE mate_goal_attempt SET lease_until=?,lease_until_epoch_second=?,updated_at=? WHERE attempt_id=? AND lease_token=? AND state IN ('claimed','running') - """, leaseUntil, now, id, leaseToken) == 1; + """, leaseUntil, GoalLeaseTime.epoch(leaseUntil), now, id, leaseToken) == 1; } public boolean checkpoint(String id, String leaseToken, String replaySafety, @@ -91,9 +98,9 @@ public class GoalAttemptStore { public List expired(LocalDateTime now, int limit) { return jdbc.query(""" SELECT * FROM mate_goal_attempt - WHERE state IN ('claimed','running') AND lease_until<=? - ORDER BY lease_until,created_at LIMIT ? - """, (rs, row) -> read(rs), now, Math.max(1, Math.min(limit, 100))); + WHERE state IN ('claimed','running') AND lease_until_epoch_second<=? + ORDER BY lease_until_epoch_second,created_at LIMIT ? + """, (rs, row) -> read(rs), GoalLeaseTime.epoch(now), Math.max(1, Math.min(limit, 100))); } private static GoalAttempt read(ResultSet rs) throws SQLException { @@ -101,7 +108,7 @@ public class GoalAttemptStore { rs.getString("attempt_id"), rs.getLong("goal_id"), rs.getString("conversation_id"), rs.getString("parent_attempt_id"), rs.getString("trigger_type"), rs.getString("state"), - rs.getString("lease_token"), time(rs, "lease_until"), + rs.getString("lease_token"), GoalLeaseTime.local(rs.getLong("lease_until_epoch_second")), nullableLong(rs, "input_item_id"), nullableLong(rs, "assistant_message_id"), rs.getString("replay_safety"), rs.getString("checkpoint_type"), rs.getString("finish_reason"), rs.getString("error_category"), diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java index 99d58a40..e2bc9493 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java @@ -18,7 +18,7 @@ public class GoalContinuationStore { """; private static final String DUE = """ ((c.state IN ('queued','retry') AND c.next_run_at<=?) - OR (c.state='running' AND c.lease_until<=?)) + OR (c.state='running' AND c.lease_until_epoch_second<=?)) """; public GoalContinuationStore(JdbcTemplate jdbc) { this.jdbc = jdbc; } @@ -63,7 +63,7 @@ public class GoalContinuationStore { SELECT c.*,g.conversation_id FROM mate_goal_continuation c JOIN mate_agent_goal g ON g.id=c.goal_id WHERE """ + ELIGIBLE + " AND " + DUE + " ORDER BY c.next_run_at,c.goal_id LIMIT ?", - (rs, row) -> read(rs), now, now, Math.max(1, Math.min(limit, 100))); + (rs, row) -> read(rs), now, GoalLeaseTime.epoch(now), Math.max(1, Math.min(limit, 100))); } public Continuation get(Long goalId) { @@ -76,20 +76,20 @@ public class GoalContinuationStore { public boolean claim(Long goalId, String token, LocalDateTime now, LocalDateTime until) { return jdbc.update(""" - UPDATE mate_goal_continuation SET state='running',lease_owner=?,lease_until=?,updated_at=?, + UPDATE mate_goal_continuation SET state='running',lease_owner=?,lease_until=?,lease_until_epoch_second=?,updated_at=?, wake_requested=FALSE,revision=revision+1 WHERE goal_id=? AND ((state IN ('queued','retry') AND next_run_at<=?) - OR (state='running' AND lease_until<=?)) + OR (state='running' AND lease_until_epoch_second<=?)) AND EXISTS(SELECT 1 FROM mate_agent_goal g WHERE g.id=goal_id AND - """ + ELIGIBLE + ")", token, until, now, goalId, now, now) == 1; + """ + ELIGIBLE + ")", token, until, GoalLeaseTime.epoch(until), now, goalId, now, GoalLeaseTime.epoch(now)) == 1; } public boolean renew(Long goalId, String token, LocalDateTime until) { return jdbc.update(""" - UPDATE mate_goal_continuation SET lease_until=? + UPDATE mate_goal_continuation SET lease_until=?,lease_until_epoch_second=? WHERE goal_id=? AND lease_owner=? AND state='running' - """, until, goalId, token) == 1; + """, until, GoalLeaseTime.epoch(until), goalId, token) == 1; } public boolean bindAttempt(Long goalId, String token, String attemptId, long expectedRevision) { @@ -100,21 +100,20 @@ public class GoalContinuationStore { """, attemptId, LocalDateTime.now(), goalId, token, expectedRevision) == 1; } - public boolean matchesFence(Long goalId, String token, String attemptId, long revision) { - Integer count=jdbc.queryForObject(""" - SELECT COUNT(*) FROM mate_goal_continuation + public boolean matchesFence(Long goalId, String token, String attemptId, long revision, LocalDateTime now) { + return jdbc.queryForList(""" + SELECT goal_id FROM mate_goal_continuation WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? - AND revision=? AND state='running' - """,Integer.class,goalId,token,attemptId,revision); - return count!=null && count==1; + AND revision=? AND state='running' AND lease_until_epoch_second>? FOR UPDATE + """,Long.class,goalId,token,attemptId,revision,GoalLeaseTime.epoch(now)).size()==1; } public boolean renewFenced(Long goalId,String token,String attemptId,long revision,LocalDateTime until) { return jdbc.update(""" - UPDATE mate_goal_continuation SET lease_until=?,updated_at=? + UPDATE mate_goal_continuation SET lease_until=?,lease_until_epoch_second=?,updated_at=? WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND revision=? AND state='running' - """,until,LocalDateTime.now(),goalId,token,attemptId,revision)==1; + """,until,GoalLeaseTime.epoch(until),LocalDateTime.now(),goalId,token,attemptId,revision)==1; } public boolean settleFenced(Long goalId,String token,String attemptId,long revision,String state, @@ -122,7 +121,7 @@ public class GoalContinuationStore { return jdbc.update(""" UPDATE mate_goal_continuation SET state=CASE WHEN ?='waiting_approval' AND wake_requested=TRUE THEN 'queued' ELSE ? END, - next_run_at=?,failures=?,reason=?,wake_requested=FALSE,lease_owner=NULL,lease_until=NULL, + next_run_at=?,failures=?,reason=?,wake_requested=FALSE,lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0, current_attempt_id=NULL,revision=revision+1,updated_at=? WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND revision=? AND state='running' """,state,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,revision)==1; @@ -133,24 +132,24 @@ public class GoalContinuationStore { return jdbc.update(""" UPDATE mate_goal_continuation SET state=?,next_run_at=?,failures=?,reason=?,wake_requested=FALSE, - lease_owner=NULL,lease_until=NULL,current_attempt_id=NULL,revision=revision+1,updated_at=? + lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0,current_attempt_id=NULL,revision=revision+1,updated_at=? WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND state='running' - AND lease_until<=? - """,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,expiredAt)==1; + AND lease_until_epoch_second<=? + """,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,GoalLeaseTime.epoch(expiredAt))==1; } public boolean settle(Long goalId, String token, String state, LocalDateTime nextRunAt, int failures, String reason) { return jdbc.update(""" UPDATE mate_goal_continuation SET state=CASE WHEN ?='waiting_approval' AND wake_requested=TRUE THEN 'queued' ELSE ? END, - next_run_at=?,failures=?,reason=?,wake_requested=FALSE,lease_owner=NULL,lease_until=NULL,updated_at=? + next_run_at=?,failures=?,reason=?,wake_requested=FALSE,lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0,updated_at=? WHERE goal_id=? AND lease_owner=? AND state='running' """, state, state, nextRunAt, failures, bounded(reason), LocalDateTime.now(), goalId, token) == 1; } public void suspendConversation(String conversationId, String reason) { jdbc.update(""" - UPDATE mate_goal_continuation SET state='paused',reason=?,lease_owner=NULL,lease_until=NULL, + UPDATE mate_goal_continuation SET state='paused',reason=?,lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0, current_attempt_id=NULL,revision=revision+1,updated_at=? WHERE goal_id IN (SELECT id FROM mate_agent_goal WHERE conversation_id=?) """, bounded(reason), LocalDateTime.now(), conversationId); @@ -159,7 +158,7 @@ public class GoalContinuationStore { public void resume(Long goalId, LocalDateTime now) { jdbc.update(""" UPDATE mate_goal_continuation SET state='queued',next_run_at=?,failures=0,reason='resumed', - lease_owner=NULL,lease_until=NULL,current_attempt_id=NULL,revision=revision+1,updated_at=? + lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0,current_attempt_id=NULL,revision=revision+1,updated_at=? WHERE goal_id=? AND state<>'running' """, now, now, goalId); } @@ -177,7 +176,7 @@ public class GoalContinuationStore { Timestamp until = rs.getTimestamp("lease_until"); return new Continuation(rs.getLong("goal_id"), rs.getString("conversation_id"), rs.getString("state"), rs.getTimestamp("next_run_at").toLocalDateTime(), rs.getString("lease_owner"), - until == null ? null : until.toLocalDateTime(), rs.getInt("failures"), rs.getString("reason"), + until == null ? null : GoalLeaseTime.local(rs.getLong("lease_until_epoch_second")), rs.getInt("failures"), rs.getString("reason"), rs.getString("current_attempt_id"),rs.getLong("revision")); } diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalLeaseTime.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalLeaseTime.java new file mode 100644 index 00000000..a3e09929 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalLeaseTime.java @@ -0,0 +1,12 @@ +package vip.mate.goal.service; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; + +/** Absolute persisted lease deadline; LocalDateTime is retained at scheduler API boundaries. */ +final class GoalLeaseTime { + private GoalLeaseTime() { } + static long epoch(LocalDateTime value) { return value.atZone(ZoneId.systemDefault()).toEpochSecond(); } + static LocalDateTime local(long epoch) { return LocalDateTime.ofInstant(Instant.ofEpochSecond(epoch), ZoneId.systemDefault()); } +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java index 95984d4a..c5a64517 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java @@ -19,10 +19,18 @@ public class GoalRunCoordinator { private final GoalAttemptStore attempts; private final GoalService goals; private final GoalProperties properties; + private final java.time.Clock clock; + @org.springframework.beans.factory.annotation.Autowired public GoalRunCoordinator(GoalContinuationStore continuations,GoalAttemptStore attempts,GoalService goals, GoalProperties properties) { + this(continuations, attempts, goals, properties, java.time.Clock.systemDefaultZone()); + } + + GoalRunCoordinator(GoalContinuationStore continuations,GoalAttemptStore attempts,GoalService goals, + GoalProperties properties,java.time.Clock clock) { this.continuations=continuations;this.attempts=attempts;this.goals=goals;this.properties=properties; + this.clock=clock; } public record ClaimedRun(GoalContinuationStore.Continuation candidate,GoalEntity goal, @@ -32,6 +40,7 @@ public class GoalRunCoordinator { public ClaimedRun claim(GoalContinuationStore.Continuation candidate,GoalEntity goal,LocalDateTime now) { if(candidate==null || goal==null || candidate.currentAttemptId()!=null) return null; if(!continuations.lockGoal(goal.getId())) return null; + now=currentTime(now); String token=UUID.randomUUID().toString(); LocalDateTime until=now.plusSeconds(LEASE_SECONDS); if(!continuations.claim(goal.getId(),token,now,until)) return null; @@ -52,24 +61,31 @@ public class GoalRunCoordinator { @Transactional public boolean markRunning(ClaimedRun run,LocalDateTime now) { if(run==null || !continuations.lockGoal(run.goal().getId())) return false; - if(!current(run)) return false; + now=currentTime(now); + if(!current(run,now)) return false; return attempts.markRunning(run.attempt().id(),run.attempt().leaseToken(),now); } @Transactional public boolean renew(ClaimedRun run,LocalDateTime now) { if(run==null || !continuations.lockGoal(run.goal().getId())) return false; + now=currentTime(now); + if(!current(run,now)) return false; LocalDateTime until=now.plusSeconds(LEASE_SECONDS); if(!continuations.renewFenced(run.goal().getId(),run.attempt().leaseToken(), run.attempt().id(),run.revision(),until)) return false; - return attempts.renew(run.attempt().id(),run.attempt().leaseToken(),until,now); + if(!attempts.renew(run.attempt().id(),run.attempt().leaseToken(),until,now)) { + throw new IllegalStateException("Goal attempt fence changed during renewal"); + } + return true; } @Transactional public boolean checkpoint(ClaimedRun run,String replaySafety,String checkpointType, Long assistantMessageId,LocalDateTime now) { if(run==null || !continuations.lockGoal(run.goal().getId())) return false; - if(!current(run)) return false; + now=currentTime(now); + if(!current(run,now)) return false; return attempts.checkpoint(run.attempt().id(),run.attempt().leaseToken(),replaySafety, checkpointType,assistantMessageId,now); } @@ -77,7 +93,8 @@ public class GoalRunCoordinator { @Transactional public boolean settle(ClaimedRun run,SegmentOutcome outcome,LocalDateTime now) { if(run==null || !continuations.lockGoal(run.goal().getId())) return false; - if(!current(run)) return false; + now=currentTime(now); + if(!current(run,now)) return false; GoalEntity fresh=goals.getById(run.goal().getId()); Settlement settlement=classify(run,outcome,fresh,now); if((outcome instanceof SegmentOutcome.Continue || outcome instanceof SegmentOutcome.Complete) @@ -93,9 +110,16 @@ public class GoalRunCoordinator { return true; } - private boolean current(ClaimedRun run) { + private LocalDateTime currentTime(LocalDateTime requested) { + // A tick timestamp captured before waiting for the goal lock cannot renew an expired owner. + LocalDateTime observed = LocalDateTime.now(clock); + return observed.isAfter(requested) ? observed : requested; + } + + private boolean current(ClaimedRun run,LocalDateTime now) { return run!=null && continuations.matchesFence(run.goal().getId(),run.attempt().leaseToken(), - run.attempt().id(),run.revision()); + run.attempt().id(),run.revision(),now) + && attempts.hasLiveFence(run.attempt().id(),run.attempt().leaseToken(),now); } private Settlement classify(ClaimedRun run,SegmentOutcome outcome,GoalEntity fresh,LocalDateTime now) { diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/ManagedGoalJsonService.java b/mateclaw-server/src/main/java/vip/mate/goal/service/ManagedGoalJsonService.java index 37ab7992..686dc897 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/ManagedGoalJsonService.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/ManagedGoalJsonService.java @@ -6,7 +6,6 @@ import org.springframework.transaction.annotation.Transactional; import vip.mate.exception.MateClawException; import vip.mate.agent.context.ChatOrigin; import java.util.Objects; -import java.time.LocalDateTime; import vip.mate.execution.evidence.service.JsonArtifactRecipe; import java.nio.charset.StandardCharsets; @@ -64,13 +63,13 @@ public class ManagedGoalJsonService { } static void verifyLease(RuntimeScope runtime) { - if (runtime.leaseUntil() != null && !runtime.leaseUntil().isAfter(LocalDateTime.now())) { + if (runtime.leaseUntil() != null && !runtime.leaseUntil().isAfter(Instant.now())) { throw failure(409, "Goal attempt lease expired during managed JSON operation"); } } record RuntimeScope(GoalJsonAcceptanceService.GoalScope goal, String producerKind, - String producerId, LocalDateTime leaseUntil) { } + String producerId, Instant leaseUntil) { } // All identity comes from server-created ToolContext, never model arguments. // Lock order: enabled user -> conversation -> goal -> continuation -> goal attempt. @@ -109,23 +108,23 @@ public class ManagedGoalJsonService { return new RuntimeScope(goal, "account-runtime", String.valueOf(userId), null); } var continuationLeases = jdbc.query(""" - SELECT lease_owner,current_attempt_id,state,lease_until FROM mate_goal_continuation WHERE goal_id=? FOR UPDATE + SELECT lease_owner,current_attempt_id,state,lease_until_epoch_second FROM mate_goal_continuation WHERE goal_id=? FOR UPDATE """, (r, i) -> Objects.equals(r.getString("lease_owner"), attribution.ownerFence()) && Objects.equals(r.getString("current_attempt_id"), attribution.goalAttemptId()) - && "running".equals(r.getString("state")) && r.getTimestamp("lease_until") != null - ? r.getTimestamp("lease_until").toLocalDateTime() : null, goal.id()); + && "running".equals(r.getString("state")) + ? Instant.ofEpochSecond(r.getLong("lease_until_epoch_second")) : null, goal.id()); if (continuationLeases.size() != 1 || continuationLeases.getFirst() == null - || !continuationLeases.getFirst().isAfter(LocalDateTime.now())) { + || !continuationLeases.getFirst().isAfter(Instant.now())) { throw failure(409, "Goal continuation owner is no longer current"); } var leases = jdbc.query(""" - SELECT goal_id,conversation_id,lease_token,state,lease_until FROM mate_goal_attempt WHERE attempt_id=? FOR UPDATE + SELECT goal_id,conversation_id,lease_token,state,lease_until_epoch_second FROM mate_goal_attempt WHERE attempt_id=? FOR UPDATE """, (r, i) -> r.getLong("goal_id") == goal.id() && Objects.equals(r.getString("conversation_id"), goal.conversationId()) && Objects.equals(r.getString("lease_token"), attribution.ownerFence()) && List.of("claimed", "running").contains(r.getString("state")) - ? r.getTimestamp("lease_until").toLocalDateTime() : null, attribution.goalAttemptId()); - if (leases.size() != 1 || leases.getFirst() == null || !leases.getFirst().isAfter(LocalDateTime.now())) { + ? Instant.ofEpochSecond(r.getLong("lease_until_epoch_second")) : null, attribution.goalAttemptId()); + if (leases.size() != 1 || leases.getFirst() == null || !leases.getFirst().isAfter(Instant.now())) { throw failure(409, "Goal attempt owner fence is no longer current"); } return new RuntimeScope(goal, "goal-attempt", attribution.goalAttemptId(), diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V198__goal_absolute_owner_leases.sql b/mateclaw-server/src/main/resources/db/migration/h2/V198__goal_absolute_owner_leases.sql new file mode 100644 index 00000000..5acca2e5 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V198__goal_absolute_owner_leases.sql @@ -0,0 +1,6 @@ +-- Unknown legacy lease timezones must not resurrect old owners on restart. +-- Zero expires existing leases; recovery retains their checkpoint/replay-safety decisions. +ALTER TABLE mate_goal_attempt ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +ALTER TABLE mate_goal_continuation ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +CREATE INDEX idx_goal_attempt_lease_epoch ON mate_goal_attempt(state, lease_until_epoch_second); +CREATE INDEX idx_goal_continuation_lease_epoch ON mate_goal_continuation(state, lease_until_epoch_second); diff --git a/mateclaw-server/src/main/resources/db/migration/kingbase/V198__goal_absolute_owner_leases.sql b/mateclaw-server/src/main/resources/db/migration/kingbase/V198__goal_absolute_owner_leases.sql new file mode 100644 index 00000000..5acca2e5 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/kingbase/V198__goal_absolute_owner_leases.sql @@ -0,0 +1,6 @@ +-- Unknown legacy lease timezones must not resurrect old owners on restart. +-- Zero expires existing leases; recovery retains their checkpoint/replay-safety decisions. +ALTER TABLE mate_goal_attempt ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +ALTER TABLE mate_goal_continuation ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +CREATE INDEX idx_goal_attempt_lease_epoch ON mate_goal_attempt(state, lease_until_epoch_second); +CREATE INDEX idx_goal_continuation_lease_epoch ON mate_goal_continuation(state, lease_until_epoch_second); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V198__goal_absolute_owner_leases.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V198__goal_absolute_owner_leases.sql new file mode 100644 index 00000000..5acca2e5 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V198__goal_absolute_owner_leases.sql @@ -0,0 +1,6 @@ +-- Unknown legacy lease timezones must not resurrect old owners on restart. +-- Zero expires existing leases; recovery retains their checkpoint/replay-safety decisions. +ALTER TABLE mate_goal_attempt ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +ALTER TABLE mate_goal_continuation ADD COLUMN lease_until_epoch_second BIGINT NOT NULL DEFAULT 0; +CREATE INDEX idx_goal_attempt_lease_epoch ON mate_goal_attempt(state, lease_until_epoch_second); +CREATE INDEX idx_goal_continuation_lease_epoch ON mate_goal_continuation(state, lease_until_epoch_second); diff --git a/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md b/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md index c4f1d8e8..9de8b1a7 100644 --- a/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md +++ b/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md @@ -48,3 +48,5 @@ Deploy the managed JSON service and all goal writers together. Stop old applicat V197 makes expiry authoritative in epoch seconds, independent of the JVM/JDBC timezone. Earlier managed records have wall-clock timestamps without a recoverable timezone, so upgrading keeps their bodies, requirements, generations and history but expires their acceptance eligibility. Publish a new version and check it under the current requirement. Existing completed goals remain historical completions; the migration does not reopen them. The 32-version quota still counts retained history. The host clock, database credentials and service host remain trusted. Run arbitrary external code without service/database credentials and outside the service's storage permissions if it is not trusted; setting a working directory or scanning paths does not provide OS isolation. Use coordinated backups of the database for requirements, bodies, pointers and bindings; workspace-file backups alone cannot restore managed acceptance. + +V198 also stores absolute scheduler lease deadlines. Existing leases expire during upgrade and are recovered from their persisted checkpoints: safe work may receive a new attempt; uncertain side effects remain blocked for review. Expired owners cannot renew, checkpoint, settle or use managed JSON tools. Renewal checks both current lease records after acquiring the goal lock, so a delayed scheduler tick cannot reuse an old timestamp to revive its owner. New valid owners can continue under the existing requirements. diff --git a/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md b/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md index 4f7199ae..aaf7a62e 100644 --- a/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md +++ b/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md @@ -48,3 +48,5 @@ V197 使用 epoch 秒作为有效期依据，不受 JVM/JDBC 时区变化影响。此前受管记录的本地时间戳无法可靠还原原始时区，因此升级保留正文、要求、generation 和历史，但使旧验收资格过期。需要重新发布版本并按当前要求检查。已经完成的 Goal 保持历史完成状态，不被重新打开；保留历史仍计入 32 个版本的配额。 宿主时钟、数据库凭据和服务宿主仍属于可信基础。若外部任意代码不可信，应在没有服务/数据库凭据、没有服务存储权限的隔离环境执行；仅设置工作目录或扫描路径不是操作系统隔离。备份应协调保存数据库中的要求、正文、指针和绑定，仅备份工作区文件无法恢复受管验收。 + +V198 同样以绝对时间保存调度租约截止。升级时旧租约失效，按已有检查点恢复：安全工作可建立新 attempt，不确定副作用仍阻断并要求核实。过期 owner 不能续租、提交检查点、结算或使用受管 JSON 工具。续租在获得 Goal 锁后检查两侧当前租约，迟到调度 tick 不能利用旧时间戳复活 owner；新的有效 owner 可按现有要求继续。 diff --git a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java index d28e1957..c6dd7570 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java @@ -313,14 +313,28 @@ class GoalJsonAcceptanceIntegrationTest { assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(wrong, "report", publication(1, "{}"))); var foreign = origin.withConversationId(goal(false).getConversationId()); assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(foreign, "report", publication(1, "{}"))); - jdbc.update("UPDATE mate_goal_attempt SET lease_until=? WHERE attempt_id=?", java.time.LocalDateTime.now().minusSeconds(1), run.attempt().id()); + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=? WHERE attempt_id=?", java.time.Instant.now().minusSeconds(1).getEpochSecond(), run.attempt().id()); assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(origin, "report", publication(1, "{}"))); - jdbc.update("UPDATE mate_goal_attempt SET lease_until=? WHERE attempt_id=?", java.time.LocalDateTime.now().plusMinutes(5), run.attempt().id()); + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=? WHERE attempt_id=?", java.time.Instant.now().plusSeconds(300).getEpochSecond(), run.attempt().id()); jdbc.update("UPDATE mate_goal_continuation SET current_attempt_id=? WHERE goal_id=?", UUID.randomUUID().toString(), goal.getId()); assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(origin, "report", publication(1, "{}"))); assertEquals(1, artifacts.list(goal.getId(), alice).getFirst().generation()); } + @ParameterizedTest + @ValueSource(strings = {"mate_goal_attempt", "mate_goal_continuation"}) + void expiredScheduledOwnerCannotRenewItsWayBackIntoJsonPublication(String table) { + GoalEntity goal = goal(true); + acceptance.configure(goal.getId(), "r", request(0, "summary"), alice); + var run = claimed(goal); + var now = java.time.LocalDateTime.now(); + jdbc.update("UPDATE " + table + " SET lease_until_epoch_second=? WHERE goal_id=?", java.time.Instant.now().minusSeconds(1).getEpochSecond(), goal.getId()); + assertFalse(coordinator.renew(run, now), "An expired owner must obtain a new fenced attempt"); + assertFalse(coordinator.checkpoint(run, "resolved", "tool_completed", null, now)); + assertFalse(coordinator.settle(run, new SegmentOutcome.Complete("stale owner"), now)); + assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(attemptOrigin(goal, run), "report", publication(0, "{}"))); + } + @Test void disabledOwnerAndIncompleteAttributionCannotUseSchedulerFallback() { GoalEntity goal = goal(true); acceptance.configure(goal.getId(), "r", request(0, "summary"), alice); @@ -567,10 +581,10 @@ class GoalJsonAcceptanceIntegrationTest { bindings.checkForRuntime(origin, "r", checkRequest(1, version)); var properties = new vip.mate.goal.config.GoalProperties(); properties.setEnabled(true); var tool = new vip.mate.tool.builtin.GoalManagementTool(goals, properties, new com.fasterxml.jackson.databind.ObjectMapper(), null); - jdbc.update("UPDATE mate_goal_attempt SET lease_until=? WHERE attempt_id=?", java.time.LocalDateTime.now().minusSeconds(1), run.attempt().id()); + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=? WHERE attempt_id=?", java.time.Instant.now().minusSeconds(1).getEpochSecond(), run.attempt().id()); assertTrue(tool.completeGoal(origin.toToolContext()).contains("error")); assertEquals(GoalStatus.ACTIVE, goals.getById(goal.getId()).getStatus()); - jdbc.update("UPDATE mate_goal_attempt SET lease_until=? WHERE attempt_id=?", java.time.LocalDateTime.now().plusSeconds(60), run.attempt().id()); + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=? WHERE attempt_id=?", java.time.Instant.now().plusSeconds(60).getEpochSecond(), run.attempt().id()); assertTrue(tool.completeGoal(origin.toToolContext()).contains("\"status\":\"completed\"")); } diff --git a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonTimezoneProcessProbe.java b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonTimezoneProcessProbe.java index 6ef5b917..3dd0ff10 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonTimezoneProcessProbe.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonTimezoneProcessProbe.java @@ -54,9 +54,49 @@ public class GoalJsonTimezoneProcessProbe { if (!bindings.state(goal, "timezone-owner").getFirst().status().equals("EXPIRED")) throw new AssertionError("Expiry fixture must initially be expired"); } } + jdbc.update("INSERT INTO mate_conversation(id,conversation_id,username,workspace_id,agent_id,create_time,update_time,deleted) VALUES (99004,'timezone-lease','timezone-owner',1,1,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,0)"); + var request = new GoalCreateRequest(); request.setConversationId("timezone-lease"); + request.setWorkspaceId(1L); request.setAgentId(1L); request.setTitle("lease"); request.setDescription("Owner restart fixture"); + request.setPersistentExecution(true); request.setAutoFollowupEnabled(true); + var goal = goals.create(request, "timezone-owner"); + requirements.configure(goal.getId(), "r", new GoalJsonAcceptanceService.ConfigureRequest(0L, "report", List.of("summary")), "timezone-owner"); + var continuations = context.getBean(GoalContinuationStore.class); + var coordinator = context.getBean(GoalRunCoordinator.class); + var now = java.time.LocalDateTime.now(); + continuations.discover(now); + var run = coordinator.claim(continuations.get(goal.getId()), goals.getById(goal.getId()), now); + if (run == null || !coordinator.markRunning(run, now)) throw new AssertionError("Owner fixture failed to claim"); + jdbc.update("UPDATE mate_goal_attempt SET lease_until=?,lease_until_epoch_second=? WHERE goal_id=?", now.minusSeconds(60), Instant.now().minusSeconds(60).getEpochSecond(), goal.getId()); + jdbc.update("UPDATE mate_goal_continuation SET lease_until=?,lease_until_epoch_second=? WHERE goal_id=?", now.minusSeconds(60), Instant.now().minusSeconds(60).getEpochSecond(), goal.getId()); + receipt.setProperty("owner.goal", String.valueOf(goal.getId())); + receipt.setProperty("owner.attempt", run.attempt().id()); + receipt.setProperty("owner.token", run.attempt().leaseToken()); + receipt.setProperty("owner.revision", String.valueOf(run.revision())); try (var output = Files.newOutputStream(file)) { receipt.store(output, "Disposable timezone fixture"); } } else { try (var input = Files.newInputStream(file)) { receipt.load(input); } + long ownerGoal = Long.parseLong(receipt.getProperty("owner.goal")); + var origin = vip.mate.agent.context.ChatOrigin.web("timezone-lease", "timezone-owner", 1L, null).withAgent(1L) + .withExecutionAttribution(new vip.mate.agent.context.ExecutionAttribution(ownerGoal, + receipt.getProperty("owner.attempt"), null, null, receipt.getProperty("owner.token"))); + boolean rejected = false; + try { artifacts.publishForRuntime(origin, "report", new ManagedGoalJsonService.PublishRequest(0L, "{}")); } + catch (vip.mate.exception.MateClawException expected) { rejected = true; } + if (!rejected) throw new AssertionError("Expired scheduler owner regained JSON publication after timezone change"); + var continuations = context.getBean(GoalContinuationStore.class); + var coordinator = context.getBean(GoalRunCoordinator.class); + var now = java.time.LocalDateTime.now(); + var oldRun = new GoalRunCoordinator.ClaimedRun(continuations.get(ownerGoal), goals.getById(ownerGoal), + context.getBean(GoalAttemptStore.class).get(receipt.getProperty("owner.attempt")), + Long.parseLong(receipt.getProperty("owner.revision"))); + if (coordinator.renew(oldRun, now)) throw new AssertionError("Expired owner renewed after timezone change"); + if (context.getBean(GoalRecoveryService.class).recoverExpired(now) != 1) throw new AssertionError("Expired owner was not recovered"); + var fresh = coordinator.claim(continuations.get(ownerGoal), goals.getById(ownerGoal), now); + if (fresh == null || !coordinator.markRunning(fresh, now)) throw new AssertionError("Recovery failed to claim a fresh owner"); + var freshOrigin = origin.withExecutionAttribution(new vip.mate.agent.context.ExecutionAttribution(ownerGoal, + fresh.attempt().id(), null, null, fresh.attempt().leaseToken())); + if (artifacts.publishForRuntime(freshOrigin, "report", new ManagedGoalJsonService.PublishRequest(0L, "{}")) + .generation() != 1) throw new AssertionError("Fresh owner cannot publish after recovery"); long expiredGoal = Long.parseLong(receipt.getProperty("expired.goal")); if (bindings.state(expiredGoal, "timezone-owner").getFirst().acceptanceEligible()) { throw new AssertionError("Previously expired JSON became eligible after host timezone changed"); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java index 853e6e0c..ec5ca755 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java @@ -25,7 +25,8 @@ class GoalAttemptStoreTest { new ResourceDatabasePopulator( new ClassPathResource("db/migration/h2/V120__agent_goal.sql"), new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"), - new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")) + new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"), + new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")) .execute(dataSource); store = new GoalAttemptStore(new JdbcTemplate(dataSource)); } diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java index 75d92dca..31ef0c13 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java @@ -23,7 +23,8 @@ class GoalContinuationStoreTest { new ResourceDatabasePopulator( new ClassPathResource("db/migration/h2/V120__agent_goal.sql"), new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"), - new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds); + new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"), + new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")).execute(ds); jdbc = new JdbcTemplate(ds); store = new GoalContinuationStore(jdbc); } @@ -103,7 +104,7 @@ class GoalContinuationStoreTest { var goals=org.mockito.Mockito.mock(GoalService.class); var runner=org.mockito.Mockito.mock(GoalSegmentRunner.class); var coordinator=new GoalRunCoordinator(new GoalContinuationStore(jdbc),new GoalAttemptStore(jdbc),goals, - new vip.mate.goal.config.GoalProperties()); + new vip.mate.goal.config.GoalProperties(),java.time.Clock.fixed(now.atZone(java.time.ZoneId.systemDefault()).toInstant(), java.time.ZoneId.systemDefault())); var recovery=org.mockito.Mockito.mock(GoalRecoveryService.class); var running=new vip.mate.agent.runtime.RunningConversationRegistry(); var streams=new vip.mate.channel.web.ChatStreamTracker(new com.fasterxml.jackson.databind.ObjectMapper()); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java index c20121fd..35dbfbcc 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java @@ -35,10 +35,11 @@ class GoalRecoveryServiceTest { ds.setURL("jdbc:h2:mem:"+ UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1"); new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V120__agent_goal.sql"), new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"), - new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds); + new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"), + new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")).execute(ds); jdbc=new JdbcTemplate(ds);attempts=new GoalAttemptStore(jdbc);continuations=new GoalContinuationStore(jdbc); inputs=new ConversationInputQueueStore(jdbc,new ObjectMapper()); - coordinator=new GoalRunCoordinator(continuations,attempts,goals,new vip.mate.goal.config.GoalProperties()); + coordinator=new GoalRunCoordinator(continuations,attempts,goals,new vip.mate.goal.config.GoalProperties(),java.time.Clock.fixed(now.atZone(java.time.ZoneId.systemDefault()).toInstant(), java.time.ZoneId.systemDefault())); recovery=new GoalRecoveryService(attempts,continuations,inputs,goals,new org.springframework.jdbc.datasource.DataSourceTransactionManager(ds)); jdbc.update(""" INSERT INTO mate_agent_goal(id,conversation_id,agent_id,workspace_id,created_by,title, @@ -97,6 +98,31 @@ class GoalRecoveryServiceTest { assertEquals(old.attempt().id(),continuations.get(1L).currentAttemptId()); } + @org.junit.jupiter.params.ParameterizedTest + @org.junit.jupiter.params.provider.ValueSource(booleans = {false, true}) + void legacyLeaseMigrationExpiresOwnersButPreservesRecoverySafety(boolean uncertain) { + var old = coordinator.claim(continuations.get(1L), goal, now); + assertTrue(coordinator.markRunning(old, now)); + if (uncertain) assertTrue(coordinator.checkpoint(old, "uncertain", "tool_started", null, now)); + // Recreate the pre-V198 schema while retaining real persisted attempts/checkpoints. + jdbc.execute("DROP INDEX idx_goal_attempt_lease_epoch"); + jdbc.execute("DROP INDEX idx_goal_continuation_lease_epoch"); + jdbc.execute("ALTER TABLE mate_goal_attempt DROP COLUMN lease_until_epoch_second"); + jdbc.execute("ALTER TABLE mate_goal_continuation DROP COLUMN lease_until_epoch_second"); + new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")) + .execute(jdbc.getDataSource()); + assertFalse(coordinator.renew(old, now.plusSeconds(1))); + assertEquals(1, recovery.recoverExpired(now.plusSeconds(1))); + assertEquals(uncertain ? "blocked" : "retryable", attempts.get(old.attempt().id()).state()); + assertEquals(uncertain ? "blocked" : "retry", continuations.get(1L).state()); + if (uncertain) verify(goals).pause(1L, "alice"); + else { + var fresh = coordinator.claim(continuations.get(1L), goal, now.plusSeconds(1)); + assertNotEquals(old.attempt().leaseToken(), fresh.attempt().leaseToken()); + assertEquals(old.attempt().id(), fresh.attempt().parentAttemptId()); + } + } + private GoalAttempt attempt(String checkpoint,String safety,Long messageId) { return new GoalAttempt("a",1L,"conv",null,"continuation","running","lease",now, null,messageId,safety,checkpoint,null,null,now,null,now,now); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java index f5fb5527..2b22e27d 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java @@ -32,9 +32,10 @@ class GoalRunCoordinatorTest { ds.setURL("jdbc:h2:mem:"+ UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1"); new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V120__agent_goal.sql"), new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"), - new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds); + new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"), + new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")).execute(ds); jdbc=new JdbcTemplate(ds);continuations=new GoalContinuationStore(jdbc);attempts=new GoalAttemptStore(jdbc); - coordinator=new GoalRunCoordinator(continuations,attempts,goals,properties); + coordinator=new GoalRunCoordinator(continuations,attempts,goals,properties,java.time.Clock.fixed(now.atZone(java.time.ZoneId.systemDefault()).toInstant(), java.time.ZoneId.systemDefault())); jdbc.update(""" INSERT INTO mate_agent_goal(id,conversation_id,agent_id,workspace_id,created_by,title, description,status,persistent_execution,auto_followup_enabled,create_time,update_time) @@ -87,6 +88,22 @@ class GoalRunCoordinatorTest { assertTrue(coordinator.settle(second,new SegmentOutcome.Continue("unfinished"),secondStart)); assertEquals(secondStart.plusSeconds(600),continuations.get(1L).nextRunAt()); } + @Test void delayedTickCannotRenewUsingItsPreLockTimestamp() { + var reference = new java.util.concurrent.atomic.AtomicReference<>(now.atZone(java.time.ZoneId.systemDefault()).toInstant()); + var clock = new java.time.Clock() { + public java.time.ZoneId getZone() { return java.time.ZoneId.systemDefault(); } + public java.time.Clock withZone(java.time.ZoneId zone) { return java.time.Clock.fixed(instant(), zone); } + public java.time.Instant instant() { return reference.get(); } + }; + var timed = new GoalRunCoordinator(continuations, attempts, goals, properties, clock); + var run = timed.claim(continuations.get(1L), goal, now); + assertTrue(timed.markRunning(run, now)); + reference.set(reference.get().plusSeconds(61)); + assertFalse(timed.renew(run, now)); + assertFalse(timed.checkpoint(run, "resolved", "tool_completed", null, now)); + assertFalse(timed.settle(run, new SegmentOutcome.Complete("delayed"), now)); + } + @Test void selectedJsonGoalCannotSettleCompletedFromSegmentClaimAlone() { goal.setJsonAcceptanceRequired(true); var run=coordinator.claim(continuations.get(1L),goal,now);