Add council review tally command
This commit is contained in:
@@ -3,8 +3,10 @@ package store
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
@@ -67,6 +69,59 @@ type CouncilWaitResult struct {
|
||||
ReviewerStatuses []CouncilReviewer `json:"reviewers"`
|
||||
}
|
||||
|
||||
type CouncilFinding struct {
|
||||
RunID string `json:"run_id"`
|
||||
ReviewerRole string `json:"reviewer_role"`
|
||||
FindingID string `json:"finding_id"`
|
||||
Title string `json:"title"`
|
||||
Summary string `json:"summary"`
|
||||
Proposal string `json:"proposal"`
|
||||
Rationale string `json:"rationale"`
|
||||
Confidence string `json:"confidence"`
|
||||
TagsJSON json.RawMessage `json:"tags_json"`
|
||||
TargetRefsJSON json.RawMessage `json:"target_refs_json"`
|
||||
}
|
||||
|
||||
type CouncilGroup struct {
|
||||
RunID string `json:"run_id"`
|
||||
GroupID string `json:"group_id"`
|
||||
Proposal string `json:"proposal"`
|
||||
Bucket string `json:"bucket"`
|
||||
SupportCount int `json:"support_count"`
|
||||
SupportersJSON json.RawMessage `json:"supporters_json"`
|
||||
DissentersJSON json.RawMessage `json:"dissenters_json"`
|
||||
RationaleSummary string `json:"rationale_summary"`
|
||||
TagsJSON json.RawMessage `json:"tags_json"`
|
||||
SourceFindingIDsJSON json.RawMessage `json:"source_finding_ids_json"`
|
||||
}
|
||||
|
||||
type CouncilTallyInput struct {
|
||||
RunID string
|
||||
Similarity string
|
||||
}
|
||||
|
||||
type CouncilTallyResult struct {
|
||||
RunID string `json:"run_id"`
|
||||
Similarity string `json:"similarity"`
|
||||
Counts map[string]int `json:"counts"`
|
||||
GroupedRecommendations []CouncilGroup `json:"grouped_recommendations"`
|
||||
}
|
||||
|
||||
type councilReviewerOutput struct {
|
||||
ReviewerRole string `json:"reviewer_role"`
|
||||
Findings []councilFindingOutput `json:"findings"`
|
||||
}
|
||||
|
||||
type councilFindingOutput struct {
|
||||
Title string `json:"title"`
|
||||
Summary string `json:"summary"`
|
||||
Proposal string `json:"proposal"`
|
||||
Rationale string `json:"rationale"`
|
||||
Confidence string `json:"confidence"`
|
||||
Tags json.RawMessage `json:"tags"`
|
||||
TargetRefs json.RawMessage `json:"target_refs"`
|
||||
}
|
||||
|
||||
func (s *OrchStore) StartCouncil(ctx context.Context, input CouncilStartInput) (CouncilStartResult, error) {
|
||||
runID := strings.TrimSpace(input.RunID)
|
||||
if runID == "" {
|
||||
@@ -621,3 +676,411 @@ func (s *OrchStore) GetCouncilReviewerStatuses(ctx context.Context, runID string
|
||||
|
||||
return reviewers, allComplete, nil
|
||||
}
|
||||
|
||||
func (s *OrchStore) TallyCouncil(ctx context.Context, input CouncilTallyInput) (CouncilTallyResult, error) {
|
||||
runID := strings.TrimSpace(input.RunID)
|
||||
if runID == "" {
|
||||
return CouncilTallyResult{}, fmt.Errorf("%w: run id is required", ErrInvalidInput)
|
||||
}
|
||||
|
||||
similarity := defaultString(strings.TrimSpace(input.Similarity), "normal")
|
||||
if similarity != "normal" && similarity != "strict" {
|
||||
return CouncilTallyResult{}, fmt.Errorf("%w: similarity must be strict or normal", ErrInvalidInput)
|
||||
}
|
||||
|
||||
if _, err := s.GetCouncilRun(ctx, runID); err != nil {
|
||||
return CouncilTallyResult{}, err
|
||||
}
|
||||
|
||||
if _, err := s.ReconcileRun(ctx, runID); err != nil && !isSQLiteBusyError(err) {
|
||||
return CouncilTallyResult{}, err
|
||||
}
|
||||
|
||||
reviewers, allComplete, err := s.GetCouncilReviewerStatuses(ctx, runID)
|
||||
if err != nil {
|
||||
return CouncilTallyResult{}, err
|
||||
}
|
||||
if !allComplete {
|
||||
return CouncilTallyResult{}, fmt.Errorf("%w: council reviewers are not complete yet", ErrInvalidState)
|
||||
}
|
||||
|
||||
findings, err := s.collectCouncilFindings(ctx, runID, reviewers)
|
||||
if err != nil {
|
||||
return CouncilTallyResult{}, err
|
||||
}
|
||||
groups := groupCouncilFindings(runID, findings, reviewers, similarity)
|
||||
counts := make(map[string]int)
|
||||
for _, group := range groups {
|
||||
counts[group.Bucket]++
|
||||
}
|
||||
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("begin council tally transaction: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM council_findings WHERE run_id = ?`, runID); err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("clear council findings: %w", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM council_groups WHERE run_id = ?`, runID); err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("clear council groups: %w", err)
|
||||
}
|
||||
|
||||
for _, finding := range findings {
|
||||
if _, err := tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO council_findings (
|
||||
run_id, reviewer_role, finding_id, title, summary, proposal, rationale,
|
||||
confidence, tags_json, target_refs_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
finding.RunID,
|
||||
finding.ReviewerRole,
|
||||
finding.FindingID,
|
||||
finding.Title,
|
||||
finding.Summary,
|
||||
finding.Proposal,
|
||||
finding.Rationale,
|
||||
finding.Confidence,
|
||||
string(finding.TagsJSON),
|
||||
string(finding.TargetRefsJSON),
|
||||
); err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("insert council finding: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
for _, group := range groups {
|
||||
if _, err := tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO council_groups (
|
||||
run_id, group_id, proposal, bucket, support_count, supporters_json,
|
||||
dissenters_json, rationale_summary, tags_json, source_finding_ids_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
group.RunID,
|
||||
group.GroupID,
|
||||
group.Proposal,
|
||||
group.Bucket,
|
||||
group.SupportCount,
|
||||
string(group.SupportersJSON),
|
||||
string(group.DissentersJSON),
|
||||
group.RationaleSummary,
|
||||
string(group.TagsJSON),
|
||||
string(group.SourceFindingIDsJSON),
|
||||
); err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("insert council group: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := insertEvent(ctx, tx, eventInput{
|
||||
RunID: runID,
|
||||
Source: "orch",
|
||||
EventType: "council_tallied",
|
||||
Summary: "council recommendations grouped",
|
||||
PayloadJSON: marshalJSON(map[string]any{
|
||||
"similarity": similarity,
|
||||
"counts": counts,
|
||||
}),
|
||||
CreatedAt: nowUTC(),
|
||||
}); err != nil {
|
||||
return CouncilTallyResult{}, err
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return CouncilTallyResult{}, fmt.Errorf("commit council tally transaction: %w", err)
|
||||
}
|
||||
|
||||
return CouncilTallyResult{
|
||||
RunID: runID,
|
||||
Similarity: similarity,
|
||||
Counts: counts,
|
||||
GroupedRecommendations: groups,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *OrchStore) collectCouncilFindings(ctx context.Context, runID string, reviewers []CouncilReviewer) ([]CouncilFinding, error) {
|
||||
findings := make([]CouncilFinding, 0)
|
||||
for _, reviewer := range reviewers {
|
||||
if reviewer.Status != "done" {
|
||||
return nil, fmt.Errorf("%w: reviewer %s did not finish successfully", ErrInvalidState, reviewer.ReviewerRole)
|
||||
}
|
||||
|
||||
message, err := s.loadCouncilReviewerResultMessage(ctx, runID, reviewer.TaskID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
output, err := parseCouncilReviewerOutput(reviewer.ReviewerRole, message)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for i, finding := range output.Findings {
|
||||
tagsJSON, err := normalizeOptionalJSONArray(finding.Tags)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: reviewer %s finding %d tags must be a JSON array", ErrInvalidInput, reviewer.ReviewerRole, i+1)
|
||||
}
|
||||
targetRefsJSON, err := normalizeOptionalJSONObject(finding.TargetRefs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: reviewer %s finding %d target_refs must be a JSON object", ErrInvalidInput, reviewer.ReviewerRole, i+1)
|
||||
}
|
||||
confidence := strings.TrimSpace(finding.Confidence)
|
||||
switch confidence {
|
||||
case "low", "medium", "high":
|
||||
default:
|
||||
return nil, fmt.Errorf("%w: reviewer %s finding %d confidence must be low, medium, or high", ErrInvalidInput, reviewer.ReviewerRole, i+1)
|
||||
}
|
||||
if strings.TrimSpace(finding.Proposal) == "" {
|
||||
return nil, fmt.Errorf("%w: reviewer %s finding %d proposal is required", ErrInvalidInput, reviewer.ReviewerRole, i+1)
|
||||
}
|
||||
findings = append(findings, CouncilFinding{
|
||||
RunID: runID,
|
||||
ReviewerRole: reviewer.ReviewerRole,
|
||||
FindingID: fmt.Sprintf("f%02d", i+1),
|
||||
Title: strings.TrimSpace(finding.Title),
|
||||
Summary: strings.TrimSpace(finding.Summary),
|
||||
Proposal: strings.TrimSpace(finding.Proposal),
|
||||
Rationale: strings.TrimSpace(finding.Rationale),
|
||||
Confidence: confidence,
|
||||
TagsJSON: json.RawMessage(tagsJSON),
|
||||
TargetRefsJSON: json.RawMessage(targetRefsJSON),
|
||||
})
|
||||
}
|
||||
}
|
||||
return findings, nil
|
||||
}
|
||||
|
||||
func (s *OrchStore) loadCouncilReviewerResultMessage(ctx context.Context, runID, taskID string) (Message, error) {
|
||||
task, err := selectTask(ctx, s.db, runID, taskID)
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
if task.LatestAttemptNo == 0 {
|
||||
return Message{}, fmt.Errorf("%w: reviewer task %s has no attempt", ErrInvalidState, taskID)
|
||||
}
|
||||
|
||||
attempt, err := selectAttempt(ctx, s.db, runID, taskID, task.LatestAttemptNo)
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
|
||||
row := s.db.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT
|
||||
message_id, thread_id, from_agent, to_agent, kind, summary, body,
|
||||
payload_json, created_at
|
||||
FROM messages
|
||||
WHERE thread_id = ? AND kind = 'result'
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1`,
|
||||
attempt.ThreadID,
|
||||
)
|
||||
message, err := scanMessage(row)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return Message{}, fmt.Errorf("%w: reviewer task %s has no result message", ErrInvalidState, taskID)
|
||||
}
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
return message, nil
|
||||
}
|
||||
|
||||
func parseCouncilReviewerOutput(expectedRole string, message Message) (councilReviewerOutput, error) {
|
||||
candidates := []string{strings.TrimSpace(message.Body), strings.TrimSpace(string(message.PayloadJSON))}
|
||||
var lastErr error
|
||||
for _, candidate := range candidates {
|
||||
if candidate == "" || candidate == "{}" {
|
||||
continue
|
||||
}
|
||||
var output councilReviewerOutput
|
||||
if err := json.Unmarshal([]byte(candidate), &output); err != nil {
|
||||
lastErr = err
|
||||
continue
|
||||
}
|
||||
if strings.TrimSpace(output.ReviewerRole) == "" {
|
||||
return councilReviewerOutput{}, fmt.Errorf("%w: reviewer output must include reviewer_role", ErrInvalidInput)
|
||||
}
|
||||
if output.ReviewerRole != expectedRole {
|
||||
return councilReviewerOutput{}, fmt.Errorf("%w: reviewer output role %s does not match expected %s", ErrInvalidInput, output.ReviewerRole, expectedRole)
|
||||
}
|
||||
return output, nil
|
||||
}
|
||||
if lastErr != nil {
|
||||
return councilReviewerOutput{}, fmt.Errorf("%w: reviewer output must be valid JSON", ErrInvalidInput)
|
||||
}
|
||||
return councilReviewerOutput{}, fmt.Errorf("%w: reviewer result message did not contain council output JSON", ErrInvalidInput)
|
||||
}
|
||||
|
||||
func normalizeOptionalJSONArray(raw json.RawMessage) (string, error) {
|
||||
if len(raw) == 0 || strings.TrimSpace(string(raw)) == "" || strings.TrimSpace(string(raw)) == "null" {
|
||||
return "[]", nil
|
||||
}
|
||||
var value []any
|
||||
if err := json.Unmarshal(raw, &value); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return marshalJSON(value), nil
|
||||
}
|
||||
|
||||
func normalizeOptionalJSONObject(raw json.RawMessage) (string, error) {
|
||||
if len(raw) == 0 || strings.TrimSpace(string(raw)) == "" || strings.TrimSpace(string(raw)) == "null" {
|
||||
return "{}", nil
|
||||
}
|
||||
var value map[string]any
|
||||
if err := json.Unmarshal(raw, &value); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return marshalJSON(value), nil
|
||||
}
|
||||
|
||||
func groupCouncilFindings(runID string, findings []CouncilFinding, reviewers []CouncilReviewer, similarity string) []CouncilGroup {
|
||||
type groupedFinding struct {
|
||||
key string
|
||||
proposal string
|
||||
findings []CouncilFinding
|
||||
}
|
||||
|
||||
order := make([]string, 0)
|
||||
groupsByKey := make(map[string]*groupedFinding)
|
||||
for _, finding := range findings {
|
||||
key := councilProposalGroupKey(finding.Proposal, similarity)
|
||||
group, ok := groupsByKey[key]
|
||||
if !ok {
|
||||
group = &groupedFinding{
|
||||
key: key,
|
||||
proposal: finding.Proposal,
|
||||
}
|
||||
groupsByKey[key] = group
|
||||
order = append(order, key)
|
||||
}
|
||||
group.findings = append(group.findings, finding)
|
||||
}
|
||||
|
||||
sortedReviewers := make([]string, 0, len(reviewers))
|
||||
for _, reviewer := range reviewers {
|
||||
sortedReviewers = append(sortedReviewers, reviewer.ReviewerRole)
|
||||
}
|
||||
sort.Strings(sortedReviewers)
|
||||
|
||||
result := make([]CouncilGroup, 0, len(order))
|
||||
for idx, key := range order {
|
||||
group := groupsByKey[key]
|
||||
supporterSet := make(map[string]struct{})
|
||||
tagSet := make(map[string]struct{})
|
||||
sourceFindingIDs := make([]string, 0, len(group.findings))
|
||||
rationaleSummary := ""
|
||||
|
||||
for _, finding := range group.findings {
|
||||
supporterSet[finding.ReviewerRole] = struct{}{}
|
||||
sourceFindingIDs = append(sourceFindingIDs, finding.ReviewerRole+":"+finding.FindingID)
|
||||
if rationaleSummary == "" && finding.Rationale != "" {
|
||||
rationaleSummary = finding.Rationale
|
||||
}
|
||||
|
||||
var tags []string
|
||||
if len(finding.TagsJSON) > 0 {
|
||||
_ = json.Unmarshal(finding.TagsJSON, &tags)
|
||||
}
|
||||
for _, tag := range tags {
|
||||
tag = strings.TrimSpace(tag)
|
||||
if tag != "" {
|
||||
tagSet[tag] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
supporters := make([]string, 0, len(supporterSet))
|
||||
for _, reviewer := range sortedReviewers {
|
||||
if _, ok := supporterSet[reviewer]; ok {
|
||||
supporters = append(supporters, reviewer)
|
||||
}
|
||||
}
|
||||
dissenters := make([]string, 0, len(sortedReviewers)-len(supporters))
|
||||
for _, reviewer := range sortedReviewers {
|
||||
if _, ok := supporterSet[reviewer]; !ok {
|
||||
dissenters = append(dissenters, reviewer)
|
||||
}
|
||||
}
|
||||
|
||||
tags := make([]string, 0, len(tagSet))
|
||||
for tag := range tagSet {
|
||||
tags = append(tags, tag)
|
||||
}
|
||||
sort.Strings(tags)
|
||||
sort.Strings(sourceFindingIDs)
|
||||
|
||||
supportCount := len(supporters)
|
||||
bucket := "minority"
|
||||
if supportCount == 3 {
|
||||
bucket = "consensus"
|
||||
} else if supportCount == 2 {
|
||||
bucket = "majority"
|
||||
}
|
||||
|
||||
result = append(result, CouncilGroup{
|
||||
RunID: runID,
|
||||
GroupID: fmt.Sprintf("grp_%02d", idx+1),
|
||||
Proposal: group.proposal,
|
||||
Bucket: bucket,
|
||||
SupportCount: supportCount,
|
||||
SupportersJSON: json.RawMessage(marshalJSON(supporters)),
|
||||
DissentersJSON: json.RawMessage(marshalJSON(dissenters)),
|
||||
RationaleSummary: rationaleSummary,
|
||||
TagsJSON: json.RawMessage(marshalJSON(tags)),
|
||||
SourceFindingIDsJSON: json.RawMessage(marshalJSON(sourceFindingIDs)),
|
||||
})
|
||||
}
|
||||
|
||||
sort.SliceStable(result, func(i, j int) bool {
|
||||
if result[i].SupportCount != result[j].SupportCount {
|
||||
return result[i].SupportCount > result[j].SupportCount
|
||||
}
|
||||
return result[i].Proposal < result[j].Proposal
|
||||
})
|
||||
for i := range result {
|
||||
result[i].GroupID = fmt.Sprintf("grp_%02d", i+1)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func councilProposalGroupKey(proposal, similarity string) string {
|
||||
tokens := proposalTokens(proposal)
|
||||
if similarity == "strict" {
|
||||
return strings.Join(tokens, " ")
|
||||
}
|
||||
|
||||
stopWords := map[string]struct{}{
|
||||
"a": {}, "an": {}, "the": {}, "to": {}, "into": {}, "and": {}, "or": {}, "of": {}, "for": {}, "in": {}, "on": {}, "with": {}, "from": {}, "that": {}, "this": {}, "it": {}, "is": {}, "are": {}, "be": {}, "by": {}, "as": {}, "keep": {}, "use": {}, "add": {},
|
||||
}
|
||||
set := make(map[string]struct{})
|
||||
filtered := make([]string, 0, len(tokens))
|
||||
for _, token := range tokens {
|
||||
if _, stop := stopWords[token]; stop {
|
||||
continue
|
||||
}
|
||||
if len(token) <= 2 {
|
||||
continue
|
||||
}
|
||||
if _, seen := set[token]; seen {
|
||||
continue
|
||||
}
|
||||
set[token] = struct{}{}
|
||||
filtered = append(filtered, token)
|
||||
}
|
||||
sort.Strings(filtered)
|
||||
if len(filtered) == 0 {
|
||||
return strings.Join(tokens, " ")
|
||||
}
|
||||
return strings.Join(filtered, " ")
|
||||
}
|
||||
|
||||
func proposalTokens(value string) []string {
|
||||
lower := strings.ToLower(strings.TrimSpace(value))
|
||||
fields := strings.FieldsFunc(lower, func(r rune) bool {
|
||||
return !((r >= 'a' && r <= 'z') || (r >= '0' && r <= '9'))
|
||||
})
|
||||
result := make([]string, 0, len(fields))
|
||||
for _, field := range fields {
|
||||
if field != "" {
|
||||
result = append(result, field)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user