mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-15 03:55:09 +08:00
fix(goal): compute owner leases from absolute clock instants
This commit is contained in:
parent
2431c1d991
commit
2c0f43c5dc
@ -23,6 +23,13 @@ public class GoalAttemptStore {
|
|||||||
public GoalAttempt create(Long goalId, String conversationId, String parentAttemptId,
|
public GoalAttempt create(Long goalId, String conversationId, String parentAttemptId,
|
||||||
String triggerType, String leaseToken, LocalDateTime leaseUntil,
|
String triggerType, String leaseToken, LocalDateTime leaseUntil,
|
||||||
Long inputItemId, LocalDateTime now) {
|
Long inputItemId, LocalDateTime now) {
|
||||||
|
return create(goalId, conversationId, parentAttemptId, triggerType, leaseToken,
|
||||||
|
leaseUntil, inputItemId, now, GoalLeaseTime.epoch(leaseUntil));
|
||||||
|
}
|
||||||
|
|
||||||
|
GoalAttempt create(Long goalId, String conversationId, String parentAttemptId,
|
||||||
|
String triggerType, String leaseToken, LocalDateTime leaseUntil,
|
||||||
|
Long inputItemId, LocalDateTime now, long leaseEpoch) {
|
||||||
String id = UUID.randomUUID().toString();
|
String id = UUID.randomUUID().toString();
|
||||||
jdbc.update("""
|
jdbc.update("""
|
||||||
INSERT INTO mate_goal_attempt(
|
INSERT INTO mate_goal_attempt(
|
||||||
@ -31,7 +38,7 @@ public class GoalAttemptStore {
|
|||||||
created_at,updated_at)
|
created_at,updated_at)
|
||||||
VALUES(?,?,?,?,?,'claimed',?,?,?,?,'safe','claimed',?,?)
|
VALUES(?,?,?,?,?,'claimed',?,?,?,?,'safe','claimed',?,?)
|
||||||
""", id, goalId, conversationId, parentAttemptId, triggerType, leaseToken,
|
""", id, goalId, conversationId, parentAttemptId, triggerType, leaseToken,
|
||||||
leaseUntil, GoalLeaseTime.epoch(leaseUntil), inputItemId, now, now);
|
leaseUntil, leaseEpoch, inputItemId, now, now);
|
||||||
return get(id);
|
return get(id);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -49,11 +56,11 @@ public class GoalAttemptStore {
|
|||||||
""", (rs, row) -> read(rs), goalId, Math.max(1, Math.min(limit, 100)));
|
""", (rs, row) -> read(rs), goalId, Math.max(1, Math.min(limit, 100)));
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean hasLiveFence(String id, String token, LocalDateTime now) {
|
public boolean hasLiveFence(String id, String token, long nowEpoch) {
|
||||||
return jdbc.queryForList("""
|
return jdbc.queryForList("""
|
||||||
SELECT attempt_id FROM mate_goal_attempt WHERE attempt_id=? AND lease_token=?
|
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
|
AND state IN ('claimed','running') AND lease_until_epoch_second>? FOR UPDATE
|
||||||
""", String.class, id, token, GoalLeaseTime.epoch(now)).size() == 1;
|
""", String.class, id, token, nowEpoch).size() == 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean markRunning(String id, String leaseToken, LocalDateTime now) {
|
public boolean markRunning(String id, String leaseToken, LocalDateTime now) {
|
||||||
@ -65,10 +72,14 @@ public class GoalAttemptStore {
|
|||||||
|
|
||||||
public boolean renew(String id, String leaseToken, LocalDateTime leaseUntil,
|
public boolean renew(String id, String leaseToken, LocalDateTime leaseUntil,
|
||||||
LocalDateTime now) {
|
LocalDateTime now) {
|
||||||
|
return renew(id, leaseToken, leaseUntil, now, GoalLeaseTime.epoch(leaseUntil));
|
||||||
|
}
|
||||||
|
|
||||||
|
boolean renew(String id, String leaseToken, LocalDateTime leaseUntil, LocalDateTime now, long leaseEpoch) {
|
||||||
return jdbc.update("""
|
return jdbc.update("""
|
||||||
UPDATE mate_goal_attempt SET lease_until=?,lease_until_epoch_second=?,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')
|
WHERE attempt_id=? AND lease_token=? AND state IN ('claimed','running')
|
||||||
""", leaseUntil, GoalLeaseTime.epoch(leaseUntil), now, id, leaseToken) == 1;
|
""", leaseUntil, leaseEpoch, now, id, leaseToken) == 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean checkpoint(String id, String leaseToken, String replaySafety,
|
public boolean checkpoint(String id, String leaseToken, String replaySafety,
|
||||||
@ -96,11 +107,15 @@ public class GoalAttemptStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public List<GoalAttempt> expired(LocalDateTime now, int limit) {
|
public List<GoalAttempt> expired(LocalDateTime now, int limit) {
|
||||||
|
return expired(GoalLeaseTime.epoch(now), limit);
|
||||||
|
}
|
||||||
|
|
||||||
|
List<GoalAttempt> expired(long nowEpoch, int limit) {
|
||||||
return jdbc.query("""
|
return jdbc.query("""
|
||||||
SELECT * FROM mate_goal_attempt
|
SELECT * FROM mate_goal_attempt
|
||||||
WHERE state IN ('claimed','running') AND lease_until_epoch_second<=?
|
WHERE state IN ('claimed','running') AND lease_until_epoch_second<=?
|
||||||
ORDER BY lease_until_epoch_second,created_at LIMIT ?
|
ORDER BY lease_until_epoch_second,created_at LIMIT ?
|
||||||
""", (rs, row) -> read(rs), GoalLeaseTime.epoch(now), Math.max(1, Math.min(limit, 100)));
|
""", (rs, row) -> read(rs), nowEpoch, Math.max(1, Math.min(limit, 100)));
|
||||||
}
|
}
|
||||||
|
|
||||||
private static GoalAttempt read(ResultSet rs) throws SQLException {
|
private static GoalAttempt read(ResultSet rs) throws SQLException {
|
||||||
|
|||||||
@ -75,6 +75,10 @@ public class GoalContinuationStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public boolean claim(Long goalId, String token, LocalDateTime now, LocalDateTime until) {
|
public boolean claim(Long goalId, String token, LocalDateTime now, LocalDateTime until) {
|
||||||
|
return claim(goalId, token, now, until, GoalLeaseTime.epoch(now), GoalLeaseTime.epoch(until));
|
||||||
|
}
|
||||||
|
|
||||||
|
boolean claim(Long goalId, String token, LocalDateTime now, LocalDateTime until, long nowEpoch, long untilEpoch) {
|
||||||
return jdbc.update("""
|
return jdbc.update("""
|
||||||
UPDATE mate_goal_continuation SET state='running',lease_owner=?,lease_until=?,lease_until_epoch_second=?,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
|
wake_requested=FALSE,revision=revision+1
|
||||||
@ -82,7 +86,7 @@ public class GoalContinuationStore {
|
|||||||
((state IN ('queued','retry') AND next_run_at<=?)
|
((state IN ('queued','retry') AND next_run_at<=?)
|
||||||
OR (state='running' AND lease_until_epoch_second<=?))
|
OR (state='running' AND lease_until_epoch_second<=?))
|
||||||
AND EXISTS(SELECT 1 FROM mate_agent_goal g WHERE g.id=goal_id AND
|
AND EXISTS(SELECT 1 FROM mate_agent_goal g WHERE g.id=goal_id AND
|
||||||
""" + ELIGIBLE + ")", token, until, GoalLeaseTime.epoch(until), now, goalId, now, GoalLeaseTime.epoch(now)) == 1;
|
""" + ELIGIBLE + ")", token, until, untilEpoch, now, goalId, now, nowEpoch) == 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean renew(Long goalId, String token, LocalDateTime until) {
|
public boolean renew(Long goalId, String token, LocalDateTime until) {
|
||||||
@ -100,20 +104,24 @@ public class GoalContinuationStore {
|
|||||||
""", attemptId, LocalDateTime.now(), goalId, token, expectedRevision) == 1;
|
""", attemptId, LocalDateTime.now(), goalId, token, expectedRevision) == 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean matchesFence(Long goalId, String token, String attemptId, long revision, LocalDateTime now) {
|
public boolean matchesFence(Long goalId, String token, String attemptId, long revision, long nowEpoch) {
|
||||||
return jdbc.queryForList("""
|
return jdbc.queryForList("""
|
||||||
SELECT goal_id FROM mate_goal_continuation
|
SELECT goal_id FROM mate_goal_continuation
|
||||||
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=?
|
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=?
|
||||||
AND revision=? AND state='running' AND lease_until_epoch_second>? FOR UPDATE
|
AND revision=? AND state='running' AND lease_until_epoch_second>? FOR UPDATE
|
||||||
""",Long.class,goalId,token,attemptId,revision,GoalLeaseTime.epoch(now)).size()==1;
|
""",Long.class,goalId,token,attemptId,revision,nowEpoch).size()==1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean renewFenced(Long goalId,String token,String attemptId,long revision,LocalDateTime until) {
|
public boolean renewFenced(Long goalId,String token,String attemptId,long revision,LocalDateTime until) {
|
||||||
|
return renewFenced(goalId, token, attemptId, revision, until, GoalLeaseTime.epoch(until));
|
||||||
|
}
|
||||||
|
|
||||||
|
boolean renewFenced(Long goalId,String token,String attemptId,long revision,LocalDateTime until,long untilEpoch) {
|
||||||
return jdbc.update("""
|
return jdbc.update("""
|
||||||
UPDATE mate_goal_continuation SET lease_until=?,lease_until_epoch_second=?,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=?
|
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=?
|
||||||
AND revision=? AND state='running'
|
AND revision=? AND state='running'
|
||||||
""",until,GoalLeaseTime.epoch(until),LocalDateTime.now(),goalId,token,attemptId,revision)==1;
|
""",until,untilEpoch,LocalDateTime.now(),goalId,token,attemptId,revision)==1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean settleFenced(Long goalId,String token,String attemptId,long revision,String state,
|
public boolean settleFenced(Long goalId,String token,String attemptId,long revision,String state,
|
||||||
@ -127,7 +135,7 @@ public class GoalContinuationStore {
|
|||||||
""",state,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,revision)==1;
|
""",state,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,revision)==1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean recoverExpired(Long goalId,String token,String attemptId,LocalDateTime expiredAt,
|
public boolean recoverExpired(Long goalId,String token,String attemptId,long expiredEpoch,
|
||||||
String state,LocalDateTime nextRunAt,int failures,String reason,LocalDateTime now) {
|
String state,LocalDateTime nextRunAt,int failures,String reason,LocalDateTime now) {
|
||||||
return jdbc.update("""
|
return jdbc.update("""
|
||||||
UPDATE mate_goal_continuation
|
UPDATE mate_goal_continuation
|
||||||
@ -135,7 +143,7 @@ public class GoalContinuationStore {
|
|||||||
lease_owner=NULL,lease_until=NULL,lease_until_epoch_second=0,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'
|
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND state='running'
|
||||||
AND lease_until_epoch_second<=?
|
AND lease_until_epoch_second<=?
|
||||||
""",state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,GoalLeaseTime.epoch(expiredAt))==1;
|
""",state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,expiredEpoch)==1;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean settle(Long goalId, String token, String state, LocalDateTime nextRunAt,
|
public boolean settle(Long goalId, String token, String state, LocalDateTime nextRunAt,
|
||||||
|
|||||||
@ -66,7 +66,7 @@ public class GoalContinuationSupervisor {
|
|||||||
public void tick() {
|
public void tick() {
|
||||||
if (closing || !properties.isEnabled() || !properties.isAllowAutoFollowup()) return;
|
if (closing || !properties.isEnabled() || !properties.isAllowAutoFollowup()) return;
|
||||||
LocalDateTime now = LocalDateTime.now(clock);
|
LocalDateTime now = LocalDateTime.now(clock);
|
||||||
recovery.recoverExpired(now);
|
recovery.recoverExpired(clock.instant());
|
||||||
active.forEach((id, claimed) -> {
|
active.forEach((id, claimed) -> {
|
||||||
GoalEntity goal = goals.getById(id);
|
GoalEntity goal = goals.getById(id);
|
||||||
boolean cancelled = goal.getStatus()==GoalStatus.PAUSED || goal.getStatus()==GoalStatus.ABANDONED
|
boolean cancelled = goal.getStatus()==GoalStatus.PAUSED || goal.getStatus()==GoalStatus.ABANDONED
|
||||||
|
|||||||
@ -44,7 +44,9 @@ public class GoalRecoveryService {
|
|||||||
return RecoveryDecision.RETRY_SAFE;
|
return RecoveryDecision.RETRY_SAFE;
|
||||||
}
|
}
|
||||||
|
|
||||||
public int recoverExpired(LocalDateTime now) {
|
public int recoverExpired(java.time.Instant moment) {
|
||||||
|
long nowEpoch=moment.getEpochSecond();
|
||||||
|
LocalDateTime now=GoalLeaseTime.local(nowEpoch);
|
||||||
if(!orphanClaimsReleased) {
|
if(!orphanClaimsReleased) {
|
||||||
synchronized(this) {
|
synchronized(this) {
|
||||||
if(!orphanClaimsReleased) {
|
if(!orphanClaimsReleased) {
|
||||||
@ -54,14 +56,14 @@ public class GoalRecoveryService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
int recovered=0;
|
int recovered=0;
|
||||||
for(GoalAttempt attempt:attempts.expired(now,100)) {
|
for(GoalAttempt attempt:attempts.expired(nowEpoch,100)) {
|
||||||
if(Boolean.TRUE.equals(transactions.execute(status -> recover(attempt,now)))) recovered++;
|
if(Boolean.TRUE.equals(transactions.execute(status -> recover(attempt,now,nowEpoch)))) recovered++;
|
||||||
}
|
}
|
||||||
return recovered;
|
return recovered;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
boolean recover(GoalAttempt attempt,LocalDateTime now) {
|
boolean recover(GoalAttempt attempt,LocalDateTime now,long nowEpoch) {
|
||||||
if(!continuations.lockGoal(attempt.goalId())) return false;
|
if(!continuations.lockGoal(attempt.goalId())) return false;
|
||||||
var continuation=continuations.get(attempt.goalId());
|
var continuation=continuations.get(attempt.goalId());
|
||||||
if(continuation==null || !attempt.id().equals(continuation.currentAttemptId())
|
if(continuation==null || !attempt.id().equals(continuation.currentAttemptId())
|
||||||
@ -73,7 +75,7 @@ public class GoalRecoveryService {
|
|||||||
? "uncertain_tool_outcome_requires_review" : "restart_recovery";
|
? "uncertain_tool_outcome_requires_review" : "restart_recovery";
|
||||||
if(!attempts.finish(attempt.id(),attempt.leaseToken(),attemptState,reason,
|
if(!attempts.finish(attempt.id(),attempt.leaseToken(),attemptState,reason,
|
||||||
decision.name().toLowerCase(),now)) return false;
|
decision.name().toLowerCase(),now)) return false;
|
||||||
if(!continuations.recoverExpired(attempt.goalId(),attempt.leaseToken(),attempt.id(),now,
|
if(!continuations.recoverExpired(attempt.goalId(),attempt.leaseToken(),attempt.id(),nowEpoch,
|
||||||
projectionState,now,continuation.failures()+1,reason,now)) {
|
projectionState,now,continuation.failures()+1,reason,now)) {
|
||||||
throw new IllegalStateException("Expired goal projection changed during recovery");
|
throw new IllegalStateException("Expired goal projection changed during recovery");
|
||||||
}
|
}
|
||||||
|
|||||||
@ -40,10 +40,13 @@ public class GoalRunCoordinator {
|
|||||||
public ClaimedRun claim(GoalContinuationStore.Continuation candidate,GoalEntity goal,LocalDateTime now) {
|
public ClaimedRun claim(GoalContinuationStore.Continuation candidate,GoalEntity goal,LocalDateTime now) {
|
||||||
if(candidate==null || goal==null || candidate.currentAttemptId()!=null) return null;
|
if(candidate==null || goal==null || candidate.currentAttemptId()!=null) return null;
|
||||||
if(!continuations.lockGoal(goal.getId())) return null;
|
if(!continuations.lockGoal(goal.getId())) return null;
|
||||||
now=currentTime(now);
|
java.time.Instant instant=currentInstant(now);
|
||||||
|
long nowEpoch=instant.getEpochSecond();
|
||||||
|
now=LocalDateTime.ofInstant(instant,java.time.ZoneId.systemDefault());
|
||||||
String token=UUID.randomUUID().toString();
|
String token=UUID.randomUUID().toString();
|
||||||
LocalDateTime until=now.plusSeconds(LEASE_SECONDS);
|
long untilEpoch=nowEpoch+LEASE_SECONDS;
|
||||||
if(!continuations.claim(goal.getId(),token,now,until)) return null;
|
LocalDateTime until=GoalLeaseTime.local(untilEpoch);
|
||||||
|
if(!continuations.claim(goal.getId(),token,now,until,nowEpoch,untilEpoch)) return null;
|
||||||
GoalContinuationStore.Continuation claimed=continuations.get(goal.getId());
|
GoalContinuationStore.Continuation claimed=continuations.get(goal.getId());
|
||||||
String parentAttemptId=null;
|
String parentAttemptId=null;
|
||||||
if("restart_recovery".equals(candidate.reason())) {
|
if("restart_recovery".equals(candidate.reason())) {
|
||||||
@ -51,7 +54,7 @@ public class GoalRunCoordinator {
|
|||||||
if(!recent.isEmpty()) parentAttemptId=recent.getFirst().id();
|
if(!recent.isEmpty()) parentAttemptId=recent.getFirst().id();
|
||||||
}
|
}
|
||||||
GoalAttempt attempt=attempts.create(goal.getId(),goal.getConversationId(),parentAttemptId,
|
GoalAttempt attempt=attempts.create(goal.getId(),goal.getConversationId(),parentAttemptId,
|
||||||
"continuation",token,until,null,now);
|
"continuation",token,until,null,now,untilEpoch);
|
||||||
if(!continuations.bindAttempt(goal.getId(),token,attempt.id(),claimed.revision())) {
|
if(!continuations.bindAttempt(goal.getId(),token,attempt.id(),claimed.revision())) {
|
||||||
throw new IllegalStateException("Goal attempt could not be bound to its continuation");
|
throw new IllegalStateException("Goal attempt could not be bound to its continuation");
|
||||||
}
|
}
|
||||||
@ -61,20 +64,25 @@ public class GoalRunCoordinator {
|
|||||||
@Transactional
|
@Transactional
|
||||||
public boolean markRunning(ClaimedRun run,LocalDateTime now) {
|
public boolean markRunning(ClaimedRun run,LocalDateTime now) {
|
||||||
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
||||||
now=currentTime(now);
|
java.time.Instant instant=currentInstant(now);
|
||||||
if(!current(run,now)) return false;
|
long nowEpoch=instant.getEpochSecond();
|
||||||
|
now=LocalDateTime.ofInstant(instant,java.time.ZoneId.systemDefault());
|
||||||
|
if(!current(run,nowEpoch)) return false;
|
||||||
return attempts.markRunning(run.attempt().id(),run.attempt().leaseToken(),now);
|
return attempts.markRunning(run.attempt().id(),run.attempt().leaseToken(),now);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
public boolean renew(ClaimedRun run,LocalDateTime now) {
|
public boolean renew(ClaimedRun run,LocalDateTime now) {
|
||||||
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
||||||
now=currentTime(now);
|
java.time.Instant instant=currentInstant(now);
|
||||||
if(!current(run,now)) return false;
|
long nowEpoch=instant.getEpochSecond();
|
||||||
LocalDateTime until=now.plusSeconds(LEASE_SECONDS);
|
now=LocalDateTime.ofInstant(instant,java.time.ZoneId.systemDefault());
|
||||||
|
if(!current(run,nowEpoch)) return false;
|
||||||
|
long untilEpoch=nowEpoch+LEASE_SECONDS;
|
||||||
|
LocalDateTime until=GoalLeaseTime.local(untilEpoch);
|
||||||
if(!continuations.renewFenced(run.goal().getId(),run.attempt().leaseToken(),
|
if(!continuations.renewFenced(run.goal().getId(),run.attempt().leaseToken(),
|
||||||
run.attempt().id(),run.revision(),until)) return false;
|
run.attempt().id(),run.revision(),until,untilEpoch)) return false;
|
||||||
if(!attempts.renew(run.attempt().id(),run.attempt().leaseToken(),until,now)) {
|
if(!attempts.renew(run.attempt().id(),run.attempt().leaseToken(),until,now,untilEpoch)) {
|
||||||
throw new IllegalStateException("Goal attempt fence changed during renewal");
|
throw new IllegalStateException("Goal attempt fence changed during renewal");
|
||||||
}
|
}
|
||||||
return true;
|
return true;
|
||||||
@ -84,8 +92,10 @@ public class GoalRunCoordinator {
|
|||||||
public boolean checkpoint(ClaimedRun run,String replaySafety,String checkpointType,
|
public boolean checkpoint(ClaimedRun run,String replaySafety,String checkpointType,
|
||||||
Long assistantMessageId,LocalDateTime now) {
|
Long assistantMessageId,LocalDateTime now) {
|
||||||
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
||||||
now=currentTime(now);
|
java.time.Instant instant=currentInstant(now);
|
||||||
if(!current(run,now)) return false;
|
long nowEpoch=instant.getEpochSecond();
|
||||||
|
now=LocalDateTime.ofInstant(instant,java.time.ZoneId.systemDefault());
|
||||||
|
if(!current(run,nowEpoch)) return false;
|
||||||
return attempts.checkpoint(run.attempt().id(),run.attempt().leaseToken(),replaySafety,
|
return attempts.checkpoint(run.attempt().id(),run.attempt().leaseToken(),replaySafety,
|
||||||
checkpointType,assistantMessageId,now);
|
checkpointType,assistantMessageId,now);
|
||||||
}
|
}
|
||||||
@ -93,8 +103,10 @@ public class GoalRunCoordinator {
|
|||||||
@Transactional
|
@Transactional
|
||||||
public boolean settle(ClaimedRun run,SegmentOutcome outcome,LocalDateTime now) {
|
public boolean settle(ClaimedRun run,SegmentOutcome outcome,LocalDateTime now) {
|
||||||
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
if(run==null || !continuations.lockGoal(run.goal().getId())) return false;
|
||||||
now=currentTime(now);
|
java.time.Instant instant=currentInstant(now);
|
||||||
if(!current(run,now)) return false;
|
long nowEpoch=instant.getEpochSecond();
|
||||||
|
now=LocalDateTime.ofInstant(instant,java.time.ZoneId.systemDefault());
|
||||||
|
if(!current(run,nowEpoch)) return false;
|
||||||
GoalEntity fresh=goals.getById(run.goal().getId());
|
GoalEntity fresh=goals.getById(run.goal().getId());
|
||||||
Settlement settlement=classify(run,outcome,fresh,now);
|
Settlement settlement=classify(run,outcome,fresh,now);
|
||||||
if((outcome instanceof SegmentOutcome.Continue || outcome instanceof SegmentOutcome.Complete)
|
if((outcome instanceof SegmentOutcome.Continue || outcome instanceof SegmentOutcome.Complete)
|
||||||
@ -110,16 +122,18 @@ public class GoalRunCoordinator {
|
|||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
private LocalDateTime currentTime(LocalDateTime requested) {
|
private java.time.Instant currentInstant(LocalDateTime requested) {
|
||||||
// A tick timestamp captured before waiting for the goal lock cannot renew an expired owner.
|
// Preserve absolute time across DST overlap and time spent waiting for the goal lock.
|
||||||
LocalDateTime observed = LocalDateTime.now(clock);
|
// Keep sub-second precision for next_run_at comparisons; only persisted leases use seconds.
|
||||||
return observed.isAfter(requested) ? observed : requested;
|
java.time.Instant observed=clock.instant();
|
||||||
|
java.time.Instant supplied=requested.atZone(java.time.ZoneId.systemDefault()).toInstant();
|
||||||
|
return observed.isAfter(supplied) ? observed : supplied;
|
||||||
}
|
}
|
||||||
|
|
||||||
private boolean current(ClaimedRun run,LocalDateTime now) {
|
private boolean current(ClaimedRun run,long nowEpoch) {
|
||||||
return run!=null && continuations.matchesFence(run.goal().getId(),run.attempt().leaseToken(),
|
return run!=null && continuations.matchesFence(run.goal().getId(),run.attempt().leaseToken(),
|
||||||
run.attempt().id(),run.revision(),now)
|
run.attempt().id(),run.revision(),nowEpoch)
|
||||||
&& attempts.hasLiveFence(run.attempt().id(),run.attempt().leaseToken(),now);
|
&& attempts.hasLiveFence(run.attempt().id(),run.attempt().leaseToken(),nowEpoch);
|
||||||
}
|
}
|
||||||
|
|
||||||
private Settlement classify(ClaimedRun run,SegmentOutcome outcome,GoalEntity fresh,LocalDateTime now) {
|
private Settlement classify(ClaimedRun run,SegmentOutcome outcome,GoalEntity fresh,LocalDateTime now) {
|
||||||
|
|||||||
@ -52,3 +52,5 @@ The host clock, database credentials and service host remain trusted. Run arbitr
|
|||||||
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.
|
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.
|
||||||
|
|
||||||
Validation snapshot (2026-09-14): the full default backend test run passed 5,249 executed tests with 33 conditional skips; the frontend passed 386 tests. The managed contract also has real compiled ReAct/Plan graph tests for account and scheduled-owner execution. Their model choices and semantic verdicts are controlled fixtures, not online-model benchmarks. Specialized integration profiles and the proprietary Kingbase engine are outside that full-default-suite claim.
|
Validation snapshot (2026-09-14): the full default backend test run passed 5,249 executed tests with 33 conditional skips; the frontend passed 386 tests. The managed contract also has real compiled ReAct/Plan graph tests for account and scheduled-owner execution. Their model choices and semantic verdicts are controlled fixtures, not online-model benchmarks. Specialized integration profiles and the proprietary Kingbase engine are outside that full-default-suite claim.
|
||||||
|
|
||||||
|
Built-in shell/code execution is not OS-isolated from the service host. Selecting JSON acceptance does not sandbox those tools, and the protocol cannot defend against host code that can access database credentials or files. Environment-name filtering and workspace path checks do not replace that isolation. Lease deadlines are calculated from absolute instants, including daylight-saving clock rollback; scheduling display fields remain local timestamps.
|
||||||
|
|||||||
@ -52,3 +52,5 @@ V197 使用 epoch 秒作为有效期依据,不受 JVM/JDBC 时区变化影响
|
|||||||
V198 同样以绝对时间保存调度租约截止。升级时旧租约失效,按已有检查点恢复:安全工作可建立新 attempt,不确定副作用仍阻断并要求核实。过期 owner 不能续租、提交检查点、结算或使用受管 JSON 工具。续租在获得 Goal 锁后检查两侧当前租约,迟到调度 tick 不能利用旧时间戳复活 owner;新的有效 owner 可按现有要求继续。
|
V198 同样以绝对时间保存调度租约截止。升级时旧租约失效,按已有检查点恢复:安全工作可建立新 attempt,不确定副作用仍阻断并要求核实。过期 owner 不能续租、提交检查点、结算或使用受管 JSON 工具。续租在获得 Goal 锁后检查两侧当前租约,迟到调度 tick 不能利用旧时间戳复活 owner;新的有效 owner 可按现有要求继续。
|
||||||
|
|
||||||
验证快照(2026-09-14):默认后端完整测试集实际通过 5,249 项、条件跳过 33 项,前端通过 386 项。受管契约还覆盖真实编译的 ReAct/Plan 图及账户/调度 owner 四种组合;模型选择和语义评价是受控夹具,不是在线模型基准。专门集成 profile 与 Kingbase 专有引擎不包含在“默认完整测试集”结论中。
|
验证快照(2026-09-14):默认后端完整测试集实际通过 5,249 项、条件跳过 33 项,前端通过 386 项。受管契约还覆盖真实编译的 ReAct/Plan 图及账户/调度 owner 四种组合;模型选择和语义评价是受控夹具,不是在线模型基准。专门集成 profile 与 Kingbase 专有引擎不包含在“默认完整测试集”结论中。
|
||||||
|
|
||||||
|
内置 shell/code 执行没有与服务宿主做操作系统隔离。选择 JSON 验收不会把这些工具变成沙箱;此协议不能抵抗能访问数据库凭据或文件的宿主代码,环境变量名称过滤和工作区路径检查也不能替代隔离。租约截止从绝对时刻计算,覆盖夏令时回拨;调度显示字段仍使用本地时间戳。
|
||||||
|
|||||||
@ -90,7 +90,7 @@ public class GoalJsonTimezoneProcessProbe {
|
|||||||
context.getBean(GoalAttemptStore.class).get(receipt.getProperty("owner.attempt")),
|
context.getBean(GoalAttemptStore.class).get(receipt.getProperty("owner.attempt")),
|
||||||
Long.parseLong(receipt.getProperty("owner.revision")));
|
Long.parseLong(receipt.getProperty("owner.revision")));
|
||||||
if (coordinator.renew(oldRun, now)) throw new AssertionError("Expired owner renewed after timezone change");
|
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");
|
if (context.getBean(GoalRecoveryService.class).recoverExpired(Instant.now()) != 1) throw new AssertionError("Expired owner was not recovered");
|
||||||
var fresh = coordinator.claim(continuations.get(ownerGoal), goals.getById(ownerGoal), now);
|
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");
|
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,
|
var freshOrigin = origin.withExecutionAttribution(new vip.mate.agent.context.ExecutionAttribution(ownerGoal,
|
||||||
|
|||||||
@ -68,7 +68,7 @@ class GoalRecoveryServiceTest {
|
|||||||
assertTrue(coordinator.markRunning(old,now));
|
assertTrue(coordinator.markRunning(old,now));
|
||||||
var queued=inputs.enqueue("conv",2L,"alice","follow up",List.of(),now);
|
var queued=inputs.enqueue("conv",2L,"alice","follow up",List.of(),now);
|
||||||
assertTrue(inputs.claimNext("conv",old.attempt().id(),now).isPresent());
|
assertTrue(inputs.claimNext("conv",old.attempt().id(),now).isPresent());
|
||||||
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61)));
|
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61).atZone(java.time.ZoneId.systemDefault()).toInstant()));
|
||||||
assertEquals("retryable",attempts.get(old.attempt().id()).state());
|
assertEquals("retryable",attempts.get(old.attempt().id()).state());
|
||||||
assertEquals("retry",continuations.get(1L).state());
|
assertEquals("retry",continuations.get(1L).state());
|
||||||
assertEquals(1,inputs.countQueued("conv"));
|
assertEquals(1,inputs.countQueued("conv"));
|
||||||
@ -81,7 +81,7 @@ class GoalRecoveryServiceTest {
|
|||||||
var old=coordinator.claim(continuations.get(1L),goal,now);
|
var old=coordinator.claim(continuations.get(1L),goal,now);
|
||||||
assertTrue(coordinator.markRunning(old,now));
|
assertTrue(coordinator.markRunning(old,now));
|
||||||
assertTrue(coordinator.checkpoint(old,"uncertain","tool_started",null,now.plusSeconds(1)));
|
assertTrue(coordinator.checkpoint(old,"uncertain","tool_started",null,now.plusSeconds(1)));
|
||||||
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61)));
|
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61).atZone(java.time.ZoneId.systemDefault()).toInstant()));
|
||||||
assertEquals("blocked",attempts.get(old.attempt().id()).state());
|
assertEquals("blocked",attempts.get(old.attempt().id()).state());
|
||||||
assertEquals("blocked",continuations.get(1L).state());
|
assertEquals("blocked",continuations.get(1L).state());
|
||||||
verify(goals).pause(1L,"alice");
|
verify(goals).pause(1L,"alice");
|
||||||
@ -92,7 +92,7 @@ class GoalRecoveryServiceTest {
|
|||||||
assertTrue(coordinator.markRunning(old,now));
|
assertTrue(coordinator.markRunning(old,now));
|
||||||
assertTrue(coordinator.checkpoint(old,"uncertain","tool_started",null,now.plusSeconds(1)));
|
assertTrue(coordinator.checkpoint(old,"uncertain","tool_started",null,now.plusSeconds(1)));
|
||||||
doThrow(new IllegalStateException("fixture pause failure")).when(goals).pause(1L,"alice");
|
doThrow(new IllegalStateException("fixture pause failure")).when(goals).pause(1L,"alice");
|
||||||
assertThrows(IllegalStateException.class, () -> recovery.recoverExpired(now.plusSeconds(61)));
|
assertThrows(IllegalStateException.class, () -> recovery.recoverExpired(now.plusSeconds(61).atZone(java.time.ZoneId.systemDefault()).toInstant()));
|
||||||
assertEquals("running",attempts.get(old.attempt().id()).state());
|
assertEquals("running",attempts.get(old.attempt().id()).state());
|
||||||
assertEquals("running",continuations.get(1L).state());
|
assertEquals("running",continuations.get(1L).state());
|
||||||
assertEquals(old.attempt().id(),continuations.get(1L).currentAttemptId());
|
assertEquals(old.attempt().id(),continuations.get(1L).currentAttemptId());
|
||||||
@ -112,7 +112,7 @@ class GoalRecoveryServiceTest {
|
|||||||
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql"))
|
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql"))
|
||||||
.execute(jdbc.getDataSource());
|
.execute(jdbc.getDataSource());
|
||||||
assertFalse(coordinator.renew(old, now.plusSeconds(1)));
|
assertFalse(coordinator.renew(old, now.plusSeconds(1)));
|
||||||
assertEquals(1, recovery.recoverExpired(now.plusSeconds(1)));
|
assertEquals(1, recovery.recoverExpired(now.plusSeconds(1).atZone(java.time.ZoneId.systemDefault()).toInstant()));
|
||||||
assertEquals(uncertain ? "blocked" : "retryable", attempts.get(old.attempt().id()).state());
|
assertEquals(uncertain ? "blocked" : "retryable", attempts.get(old.attempt().id()).state());
|
||||||
assertEquals(uncertain ? "blocked" : "retry", continuations.get(1L).state());
|
assertEquals(uncertain ? "blocked" : "retry", continuations.get(1L).state());
|
||||||
if (uncertain) verify(goals).pause(1L, "alice");
|
if (uncertain) verify(goals).pause(1L, "alice");
|
||||||
|
|||||||
@ -17,6 +17,7 @@ import java.util.UUID;
|
|||||||
import static org.junit.jupiter.api.Assertions.*;
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
import static org.mockito.Mockito.*;
|
import static org.mockito.Mockito.*;
|
||||||
|
|
||||||
|
@org.junit.jupiter.api.parallel.Isolated
|
||||||
class GoalRunCoordinatorTest {
|
class GoalRunCoordinatorTest {
|
||||||
JdbcTemplate jdbc;
|
JdbcTemplate jdbc;
|
||||||
GoalContinuationStore continuations;
|
GoalContinuationStore continuations;
|
||||||
@ -88,6 +89,43 @@ class GoalRunCoordinatorTest {
|
|||||||
assertTrue(coordinator.settle(second,new SegmentOutcome.Continue("unfinished"),secondStart));
|
assertTrue(coordinator.settle(second,new SegmentOutcome.Continue("unfinished"),secondStart));
|
||||||
assertEquals(secondStart.plusSeconds(600),continuations.get(1L).nextRunAt());
|
assertEquals(secondStart.plusSeconds(600),continuations.get(1L).nextRunAt());
|
||||||
}
|
}
|
||||||
|
@org.junit.jupiter.params.ParameterizedTest
|
||||||
|
@org.junit.jupiter.params.provider.ValueSource(strings = {"2026-11-01T05:59:50Z", "2026-11-01T06:00:10Z"})
|
||||||
|
void leaseDurationStaysSixtyRealSecondsAcrossDstRollback(String timestamp) {
|
||||||
|
java.util.TimeZone previous = java.util.TimeZone.getDefault();
|
||||||
|
try {
|
||||||
|
java.util.TimeZone.setDefault(java.util.TimeZone.getTimeZone("America/New_York"));
|
||||||
|
var instant = java.time.Instant.parse(timestamp);
|
||||||
|
var reference = new java.util.concurrent.atomic.AtomicReference<>(instant);
|
||||||
|
var clock = new java.time.Clock() {
|
||||||
|
public java.time.ZoneId getZone() { return java.time.ZoneId.of("America/New_York"); }
|
||||||
|
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 local = java.time.LocalDateTime.ofInstant(instant, clock.getZone());
|
||||||
|
var run = timed.claim(continuations.get(1L), goal, local);
|
||||||
|
assertNotNull(run);
|
||||||
|
assertEquals(instant.plusSeconds(60).getEpochSecond(), jdbc.queryForObject(
|
||||||
|
"SELECT lease_until_epoch_second FROM mate_goal_attempt WHERE attempt_id=?", Long.class, run.attempt().id()));
|
||||||
|
reference.set(instant.plusSeconds(10));
|
||||||
|
assertTrue(timed.renew(run, java.time.LocalDateTime.ofInstant(reference.get(), clock.getZone())));
|
||||||
|
assertEquals(reference.get().plusSeconds(60).getEpochSecond(), jdbc.queryForObject(
|
||||||
|
"SELECT lease_until_epoch_second FROM mate_goal_continuation WHERE goal_id=1", Long.class));
|
||||||
|
reference.set(reference.get().plusSeconds(61));
|
||||||
|
var expiredLocal = java.time.LocalDateTime.ofInstant(reference.get(), clock.getZone());
|
||||||
|
assertFalse(timed.renew(run, expiredLocal));
|
||||||
|
var recovery = new GoalRecoveryService(attempts, continuations,
|
||||||
|
new vip.mate.channel.web.ConversationInputQueueStore(jdbc, new com.fasterxml.jackson.databind.ObjectMapper()), goals,
|
||||||
|
new org.springframework.jdbc.datasource.DataSourceTransactionManager(jdbc.getDataSource()));
|
||||||
|
assertEquals(1, recovery.recoverExpired(reference.get()));
|
||||||
|
assertEquals("retry", continuations.get(1L).state());
|
||||||
|
var next = timed.claim(continuations.get(1L), goal, expiredLocal);
|
||||||
|
assertNotNull(next);
|
||||||
|
assertNotEquals(run.attempt().leaseToken(), next.attempt().leaseToken());
|
||||||
|
} finally { java.util.TimeZone.setDefault(previous); }
|
||||||
|
}
|
||||||
|
|
||||||
@Test void delayedTickCannotRenewUsingItsPreLockTimestamp() {
|
@Test void delayedTickCannotRenewUsingItsPreLockTimestamp() {
|
||||||
var reference = new java.util.concurrent.atomic.AtomicReference<>(now.atZone(java.time.ZoneId.systemDefault()).toInstant());
|
var reference = new java.util.concurrent.atomic.AtomicReference<>(now.atZone(java.time.ZoneId.systemDefault()).toInstant());
|
||||||
var clock = new java.time.Clock() {
|
var clock = new java.time.Clock() {
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user